Skip to content

Commit 197b0fc

Browse files
affandarCopilot
affandar
and
Copilot
committed
fix: eliminate double-serialization in race() and all() results (#59)
race() and all() were double-encoding values by embedding JSON strings consumers to need extra json.loads() calls to access structured data. - Add wrap_ok/wrap_err helpers that parse values before embedding - Fix make_select_future and make_join_future for all task types - Update race/all/e2e tests to assert direct values (no extra parse) - Add type preservation tests (string, number, object, array, null, bool) - Add all()+dequeue_event and race()+dequeue_event regression tests - Add SDK symmetry section to copilot-instructions.md Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
1 parent 93a35b5 commit 197b0fc

4 files changed

Lines changed: 233 additions & 103 deletions

File tree

‎.github/copilot-instructions.md‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,18 @@
11
# duroxide-python — Copilot Instructions
22

3+
> **🚨 STOP: DO NOT `git commit` OR `git push` WITHOUT EXPLICIT USER PERMISSION 🚨**
4+
>
5+
> Always ask before committing or pushing. Never assume.
6+
7+
## ⚠️ SDK Symmetry — duroxide-python & duroxide-node MUST stay in sync
8+
9+
When making changes to this SDK, **always** check `../duroxide-node` for symmetry:
10+
1. **Bug fixes**: If you fix a bug here, apply the same fix in duroxide-node (and vice versa)
11+
2. **API parity**: Every public API in one SDK must have an equivalent in the other (adjusted for language idioms — snake_case in Python, camelCase in JS)
12+
3. **Test parity**: Every test in one SDK must have a corresponding test in the other
13+
4. **Feature parity**: New duroxide features exposed in one SDK must be exposed in both
14+
5. **Rust handler code** (`src/handlers.rs`): Both SDKs share the same interop architecture — `execute_task`, `make_select_future`, `make_join_future` should be kept structurally identical
15+
316
## Project Overview
417

518
**duroxide-python** is a Python SDK for the duroxide durable execution runtime. It wraps the Rust duroxide library via PyO3/maturin, providing a generator-based API for writing deterministic, replayable workflows.

‎src/handlers.rs‎

Lines changed: 48 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -457,12 +457,22 @@ impl PyOrchestrationHandler {
457457
let f2 = make_select_future(ctx, t2);
458458

459459
match ctx.select2(f1, f2).await {
460-
duroxide::Either2::First(val) => TaskResult::Ok(
461-
serde_json::json!({ "index": 0, "value": val }).to_string(),
462-
),
463-
duroxide::Either2::Second(val) => TaskResult::Ok(
464-
serde_json::json!({ "index": 1, "value": val }).to_string(),
465-
),
460+
duroxide::Either2::First(val) => {
461+
// Parse val as JSON so it's embedded as a structured value,
462+
// not double-serialized as a JSON string.
463+
let parsed = serde_json::from_str::<serde_json::Value>(&val)
464+
.unwrap_or(serde_json::Value::String(val));
465+
TaskResult::Ok(
466+
serde_json::json!({ "index": 0, "value": parsed }).to_string(),
467+
)
468+
}
469+
duroxide::Either2::Second(val) => {
470+
let parsed = serde_json::from_str::<serde_json::Value>(&val)
471+
.unwrap_or(serde_json::Value::String(val));
472+
TaskResult::Ok(
473+
serde_json::json!({ "index": 1, "value": parsed }).to_string(),
474+
)
475+
}
466476
}
467477
}
468478
}
@@ -588,6 +598,20 @@ fn make_select_future(
588598
}
589599
}
590600

601+
/// Wrap a raw string value as `{"ok": <parsed>}` or `{"err": <parsed>}` JSON.
602+
/// Parses the string as JSON first to avoid double-serialization.
603+
fn wrap_ok(val: String) -> String {
604+
let parsed = serde_json::from_str::<serde_json::Value>(&val)
605+
.unwrap_or(serde_json::Value::String(val));
606+
serde_json::json!({ "ok": parsed }).to_string()
607+
}
608+
609+
fn wrap_err(val: String) -> String {
610+
let parsed = serde_json::from_str::<serde_json::Value>(&val)
611+
.unwrap_or(serde_json::Value::String(val));
612+
serde_json::json!({ "err": parsed }).to_string()
613+
}
614+
591615
/// Convert a ScheduledTask into a type-erased future returning `{ok:v}/{err:e}` JSON for use in join.
592616
fn make_join_future(
593617
ctx: &OrchestrationContext,
@@ -596,8 +620,8 @@ fn make_join_future(
596620
match task {
597621
ScheduledTask::Activity { name, input } => Box::pin(async move {
598622
match ctx.schedule_activity(&name, input).await {
599-
Ok(v) => serde_json::json!({ "ok": v }).to_string(),
600-
Err(e) => serde_json::json!({ "err": e }).to_string(),
623+
Ok(v) => wrap_ok(v),
624+
Err(e) => wrap_err(e),
601625
}
602626
}),
603627
ScheduledTask::ActivityWithSession {
@@ -609,8 +633,8 @@ fn make_join_future(
609633
.schedule_activity_on_session(&name, input, session_id)
610634
.await
611635
{
612-
Ok(v) => serde_json::json!({ "ok": v }).to_string(),
613-
Err(e) => serde_json::json!({ "err": e }).to_string(),
636+
Ok(v) => wrap_ok(v),
637+
Err(e) => wrap_err(e),
614638
}
615639
}),
616640
ScheduledTask::ActivityWithRetry {
@@ -623,8 +647,8 @@ fn make_join_future(
623647
.schedule_activity_with_retry(&name, input, policy)
624648
.await
625649
{
626-
Ok(v) => serde_json::json!({ "ok": v }).to_string(),
627-
Err(e) => serde_json::json!({ "err": e }).to_string(),
650+
Ok(v) => wrap_ok(v),
651+
Err(e) => wrap_err(e),
628652
}
629653
}),
630654
ScheduledTask::Timer { delay_ms } => Box::pin(async move {
@@ -633,12 +657,12 @@ fn make_join_future(
633657
}),
634658
ScheduledTask::WaitEvent { name } => Box::pin(async move {
635659
let data = ctx.schedule_wait(&name).await;
636-
serde_json::json!({ "ok": data }).to_string()
660+
wrap_ok(data)
637661
}),
638662
ScheduledTask::SubOrchestration { name, input } => Box::pin(async move {
639663
match ctx.schedule_sub_orchestration(&name, input).await {
640-
Ok(v) => serde_json::json!({ "ok": v }).to_string(),
641-
Err(e) => serde_json::json!({ "err": e }).to_string(),
664+
Ok(v) => wrap_ok(v),
665+
Err(e) => wrap_err(e),
642666
}
643667
}),
644668
ScheduledTask::SubOrchestrationWithId {
@@ -650,8 +674,8 @@ fn make_join_future(
650674
.schedule_sub_orchestration_with_id(&name, instance_id, input)
651675
.await
652676
{
653-
Ok(v) => serde_json::json!({ "ok": v }).to_string(),
654-
Err(e) => serde_json::json!({ "err": e }).to_string(),
677+
Ok(v) => wrap_ok(v),
678+
Err(e) => wrap_err(e),
655679
}
656680
}),
657681
ScheduledTask::SubOrchestrationVersioned {
@@ -663,8 +687,8 @@ fn make_join_future(
663687
.schedule_sub_orchestration_versioned(&name, version, input)
664688
.await
665689
{
666-
Ok(v) => serde_json::json!({ "ok": v }).to_string(),
667-
Err(e) => serde_json::json!({ "err": e }).to_string(),
690+
Ok(v) => wrap_ok(v),
691+
Err(e) => wrap_err(e),
668692
}
669693
}),
670694
ScheduledTask::SubOrchestrationVersionedWithId {
@@ -677,13 +701,13 @@ fn make_join_future(
677701
.schedule_sub_orchestration_versioned_with_id(&name, version, instance_id, input)
678702
.await
679703
{
680-
Ok(v) => serde_json::json!({ "ok": v }).to_string(),
681-
Err(e) => serde_json::json!({ "err": e }).to_string(),
704+
Ok(v) => wrap_ok(v),
705+
Err(e) => wrap_err(e),
682706
}
683707
}),
684708
ScheduledTask::DequeueEvent { queue_name } => Box::pin(async move {
685709
let data = ctx.dequeue_event(&queue_name).await;
686-
serde_json::json!({ "ok": data }).to_string()
710+
wrap_ok(data)
687711
}),
688712
ScheduledTask::ActivityWithRetryOnSession {
689713
name,
@@ -696,8 +720,8 @@ fn make_join_future(
696720
.schedule_activity_with_retry_on_session(&name, input, policy, &session_id)
697721
.await
698722
{
699-
Ok(v) => serde_json::json!({ "ok": v }).to_string(),
700-
Err(e) => serde_json::json!({ "err": e }).to_string(),
723+
Ok(v) => wrap_ok(v),
724+
Err(e) => wrap_err(e),
701725
}
702726
}),
703727
_ => Box::pin(async {

‎tests/test_e2e.py‎

Lines changed: 5 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -188,15 +188,7 @@ def fan_out(ctx, input):
188188
a = ctx.schedule_activity("Greetings", "Gabbar")
189189
b = ctx.schedule_activity("Greetings", "Samba")
190190
results = yield ctx.all([a, b])
191-
vals = []
192-
for r in results:
193-
v = r.get("ok", r.get("err"))
194-
if isinstance(v, str):
195-
try:
196-
v = json.loads(v)
197-
except (json.JSONDecodeError, TypeError):
198-
pass
199-
vals.append(v)
191+
vals = [r.get("ok", r.get("err")) for r in results]
200192
vals.sort()
201193
return f"{vals[0]}, {vals[1]}"
202194

@@ -263,13 +255,7 @@ def child_upper(ctx, input):
263255
@rt.register_orchestration("Parent")
264256
def parent(ctx, input):
265257
r = yield ctx.schedule_sub_orchestration("ChildUpper", input)
266-
val = r
267-
if isinstance(val, str):
268-
try:
269-
val = json.loads(val)
270-
except (json.JSONDecodeError, TypeError):
271-
pass
272-
return f"parent:{val}"
258+
return f"parent:{r}"
273259

274260
result = run_orchestration(provider, "Parent", "hi", setup)
275261
assert result.status == "Completed"
@@ -298,11 +284,6 @@ def parent_fan(ctx, input):
298284
total = 0
299285
for r in results:
300286
v = r.get("ok", r.get("err"))
301-
if isinstance(v, str):
302-
try:
303-
v = json.loads(v)
304-
except (json.JSONDecodeError, TypeError):
305-
pass
306287
if isinstance(v, str):
307288
v = int(v)
308289
total += v
@@ -327,24 +308,12 @@ def leaf(ctx, input):
327308
@rt.register_orchestration("Mid")
328309
def mid(ctx, input):
329310
r = yield ctx.schedule_sub_orchestration("Leaf", input)
330-
val = r
331-
if isinstance(val, str):
332-
try:
333-
val = json.loads(val)
334-
except (json.JSONDecodeError, TypeError):
335-
pass
336-
return f"{val}-mid"
311+
return f"{r}-mid"
337312

338313
@rt.register_orchestration("Root")
339314
def root(ctx, input):
340315
r = yield ctx.schedule_sub_orchestration("Mid", input)
341-
val = r
342-
if isinstance(val, str):
343-
try:
344-
val = json.loads(val)
345-
except (json.JSONDecodeError, TypeError):
346-
pass
347-
return f"root:{val}"
316+
return f"root:{r}"
348317

349318
result = run_orchestration(provider, "Root", "a", setup)
350319
assert result.status == "Completed"
@@ -573,14 +542,7 @@ def join_timer(ctx, input):
573542

574543
result = run_orchestration(provider, "JoinTimer", None, setup)
575544
assert result.status == "Completed"
576-
# activity result is JSON-encoded string
577-
activity_val = result.output["activity"]
578-
if isinstance(activity_val, str):
579-
try:
580-
activity_val = json.loads(activity_val)
581-
except (json.JSONDecodeError, TypeError):
582-
pass
583-
assert activity_val == "task_done"
545+
assert result.output["activity"] == "task_done"
584546
assert result.output["timer"] is None
585547

586548

0 commit comments

Comments
 (0)