zzstoatzz.io
prefect server in zig
implement empirical_policy mutation during retry transitions
- add policy mutation flags to RuleContext (set_retry_type_in_process,
clear_retry_type, clear_pause_keys, set_resuming_false)
- RetryFailedFlows sets flags on retry: retry_type="in_process",
resuming=false, pause_keys=[]
- RetryFailedFlows clears retry_type when retries exhausted
- API handler builds updated policy JSON from flags
- setStateWithSchedule accepts optional empirical_policy update
- add tests for policy mutation flags
matches python prefect server behavior for retry policy handling.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
+10
-8
docs/python-reference/zig-compat-notes.md
+10
-8
docs/python-reference/zig-compat-notes.md
···
15
15
| retry state | `AwaitingRetry` (SCHEDULED) | `SCHEDULED` with name "AwaitingRetry" |
16
16
| response | `reject_transition(state, reason="Retrying")` | `REJECT` status |
17
17
18
-
### gaps to address later
18
+
### implemented ✓
19
19
20
-
1. **empirical_policy updates during retry**
21
-
- python sets `retry_type = "in_process"`, `resuming = False`, `pause_keys = set()` on retry
22
-
- python clears `retry_type = None` when retries exhausted
23
-
- zig: not yet implemented (policy stored but not mutated)
20
+
| behavior | status |
21
+
|----------|--------|
22
+
| retry_type = "in_process" on retry | ✓ via `set_retry_type_in_process` flag |
23
+
| resuming = false on retry | ✓ via `set_resuming_false` flag |
24
+
| pause_keys = [] on retry | ✓ via `clear_pause_keys` flag |
25
+
| retry_type = null when exhausted | ✓ via `clear_retry_type` flag |
26
+
27
+
### not needed
24
28
25
-
2. **old client compatibility** (API < 0.8.3)
26
-
- python resets failed task runs to AwaitingRetry state
27
-
- zig: not needed for new clients
29
+
- **old client compatibility** (API < 0.8.3): python resets failed task runs to AwaitingRetry state - not needed for new clients
28
30
29
31
## schema compatibility
30
32
+2
-2
loq.toml
+2
-2
loq.toml
+149
-5
src/api/flow_runs.zig
+149
-5
src/api/flow_runs.zig
···
2
2
const zap = @import("zap");
3
3
const mem = std.mem;
4
4
const json = std.json;
5
+
const Allocator = std.mem.Allocator;
5
6
6
7
const db = @import("../db/sqlite.zig");
7
8
const routing = @import("routing.zig");
···
9
10
const time_util = @import("../utilities/time.zig");
10
11
const json_util = @import("../utilities/json.zig");
11
12
const orchestration = @import("../orchestration.zig");
13
+
14
+
/// Build updated empirical_policy JSON based on rule context flags
15
+
/// Preserves existing fields from current_policy while applying updates
16
+
fn buildUpdatedPolicy(
17
+
alloc: Allocator,
18
+
current_policy: []const u8,
19
+
rule_ctx: *const orchestration.RuleContext,
20
+
) ?[]const u8 {
21
+
// check if any policy updates are needed
22
+
if (!rule_ctx.set_retry_type_in_process and
23
+
!rule_ctx.clear_retry_type and
24
+
!rule_ctx.clear_pause_keys and
25
+
!rule_ctx.set_resuming_false)
26
+
{
27
+
return null;
28
+
}
29
+
30
+
// parse current policy
31
+
const parsed = json.parseFromSlice(json.Value, alloc, current_policy, .{}) catch {
32
+
// if parse fails, start with empty object
33
+
return buildFreshPolicy(alloc, rule_ctx);
34
+
};
35
+
const current = parsed.value;
36
+
37
+
if (current != .object) {
38
+
return buildFreshPolicy(alloc, rule_ctx);
39
+
}
40
+
41
+
// build new policy with updates
42
+
var out: std.Io.Writer.Allocating = .init(alloc);
43
+
var jw: json.Stringify = .{ .writer = &out.writer };
44
+
45
+
jw.beginObject() catch return null;
46
+
47
+
// copy existing fields, applying updates
48
+
var it = current.object.iterator();
49
+
while (it.next()) |entry| {
50
+
const key = entry.key_ptr.*;
51
+
52
+
// skip fields we're updating
53
+
if (mem.eql(u8, key, "retry_type") or
54
+
mem.eql(u8, key, "pause_keys") or
55
+
mem.eql(u8, key, "resuming"))
56
+
{
57
+
continue;
58
+
}
59
+
60
+
jw.objectField(key) catch return null;
61
+
jw.write(entry.value_ptr.*) catch return null;
62
+
}
63
+
64
+
// apply updates
65
+
if (rule_ctx.set_retry_type_in_process) {
66
+
jw.objectField("retry_type") catch return null;
67
+
jw.write("in_process") catch return null;
68
+
} else if (rule_ctx.clear_retry_type) {
69
+
jw.objectField("retry_type") catch return null;
70
+
jw.write(null) catch return null;
71
+
}
72
+
73
+
if (rule_ctx.clear_pause_keys) {
74
+
jw.objectField("pause_keys") catch return null;
75
+
jw.beginArray() catch return null;
76
+
jw.endArray() catch return null;
77
+
}
78
+
79
+
if (rule_ctx.set_resuming_false) {
80
+
jw.objectField("resuming") catch return null;
81
+
jw.write(false) catch return null;
82
+
}
83
+
84
+
jw.endObject() catch return null;
85
+
return out.toOwnedSlice() catch null;
86
+
}
87
+
88
+
/// Build a fresh policy when current is invalid/empty
89
+
fn buildFreshPolicy(alloc: Allocator, rule_ctx: *const orchestration.RuleContext) ?[]const u8 {
90
+
var out: std.Io.Writer.Allocating = .init(alloc);
91
+
var jw: json.Stringify = .{ .writer = &out.writer };
92
+
93
+
jw.beginObject() catch return null;
94
+
95
+
if (rule_ctx.retries) |r| {
96
+
jw.objectField("retries") catch return null;
97
+
jw.write(r) catch return null;
98
+
}
99
+
if (rule_ctx.retry_delay) |d| {
100
+
jw.objectField("retry_delay") catch return null;
101
+
jw.write(d) catch return null;
102
+
}
103
+
104
+
if (rule_ctx.set_retry_type_in_process) {
105
+
jw.objectField("retry_type") catch return null;
106
+
jw.write("in_process") catch return null;
107
+
} else if (rule_ctx.clear_retry_type) {
108
+
jw.objectField("retry_type") catch return null;
109
+
jw.write(null) catch return null;
110
+
}
111
+
112
+
if (rule_ctx.clear_pause_keys) {
113
+
jw.objectField("pause_keys") catch return null;
114
+
jw.beginArray() catch return null;
115
+
jw.endArray() catch return null;
116
+
}
117
+
118
+
if (rule_ctx.set_resuming_false) {
119
+
jw.objectField("resuming") catch return null;
120
+
jw.write(false) catch return null;
121
+
}
122
+
123
+
jw.endObject() catch return null;
124
+
return out.toOwnedSlice() catch null;
125
+
}
12
126
13
127
// POST /flow_runs/ - create flow run
14
128
// GET /flow_runs/{id} - read flow run
···
329
443
};
330
444
orchestration.applyBookkeeping(&bookkeeping_ctx);
331
445
332
-
// commit the retry state with next_scheduled_start_time
446
+
// build updated empirical_policy if rule set mutation flags
447
+
const current_policy = if (current_run) |run| run.empirical_policy else "{}";
448
+
const updated_policy = buildUpdatedPolicy(alloc, current_policy, &rule_ctx);
449
+
450
+
// commit the retry state with next_scheduled_start_time and updated policy
333
451
db.setFlowRunStateWithSchedule(
334
452
id,
335
453
state_id,
···
342
460
bookkeeping_ctx.new_total_run_time,
343
461
null, // expected_start_time
344
462
retry_scheduled, // next_scheduled_start_time
463
+
updated_policy, // empirical_policy updates
345
464
) catch {
346
465
json_util.sendStatus(r, "{\"detail\":\"update failed\"}", .internal_server_error);
347
466
return;
···
399
518
};
400
519
orchestration.applyBookkeeping(&bookkeeping_ctx);
401
520
521
+
// check if policy needs updating (e.g., clear_retry_type when retries exhausted)
522
+
const current_policy = if (current_run) |run| run.empirical_policy else "{}";
523
+
const updated_policy = buildUpdatedPolicy(alloc, current_policy, &rule_ctx);
524
+
402
525
// atomic state transition with orchestration data
403
-
db.setFlowRunState(id, state_id, state_type, state_name, now, bookkeeping_ctx.new_start_time, bookkeeping_ctx.new_end_time, bookkeeping_ctx.new_run_count, bookkeeping_ctx.new_total_run_time, rule_ctx.new_expected_start_time) catch {
404
-
json_util.sendStatus(r, "{\"detail\":\"update failed\"}", .internal_server_error);
405
-
return;
406
-
};
526
+
if (updated_policy != null) {
527
+
// use setStateWithSchedule to update policy (pass null for next_scheduled_start_time)
528
+
db.setFlowRunStateWithSchedule(
529
+
id,
530
+
state_id,
531
+
state_type,
532
+
state_name,
533
+
now,
534
+
bookkeeping_ctx.new_start_time,
535
+
bookkeeping_ctx.new_end_time,
536
+
bookkeeping_ctx.new_run_count,
537
+
bookkeeping_ctx.new_total_run_time,
538
+
rule_ctx.new_expected_start_time,
539
+
null, // next_scheduled_start_time
540
+
updated_policy,
541
+
) catch {
542
+
json_util.sendStatus(r, "{\"detail\":\"update failed\"}", .internal_server_error);
543
+
return;
544
+
};
545
+
} else {
546
+
db.setFlowRunState(id, state_id, state_type, state_name, now, bookkeeping_ctx.new_start_time, bookkeeping_ctx.new_end_time, bookkeeping_ctx.new_run_count, bookkeeping_ctx.new_total_run_time, rule_ctx.new_expected_start_time) catch {
547
+
json_util.sendStatus(r, "{\"detail\":\"update failed\"}", .internal_server_error);
548
+
return;
549
+
};
550
+
}
407
551
408
552
const resp = writeStateResponse(alloc, .ACCEPT, .{}, state_type, state_name, now, state_id) catch {
409
553
json_util.sendStatus(r, "{\"detail\":\"serialize error\"}", .internal_server_error);
+53
-23
src/db/flow_runs.zig
+53
-23
src/db/flow_runs.zig
···
221
221
}
222
222
223
223
/// Set flow run state with next_scheduled_start_time (for AwaitingRetry state)
224
+
/// optionally updates empirical_policy if provided
224
225
pub fn setStateWithSchedule(
225
226
run_id: []const u8,
226
227
state_id: []const u8,
···
233
234
total_run_time: f64,
234
235
expected_start_time: ?[]const u8,
235
236
next_scheduled_start_time: ?[]const u8,
237
+
empirical_policy: ?[]const u8,
236
238
) !void {
237
239
if (backend.db.dialect == .sqlite) {
238
240
backend.db.mutex.lock();
···
247
249
};
248
250
errdefer txn.rollback();
249
251
250
-
txn.exec(
251
-
\\UPDATE flow_run SET state_id = ?, state_type = ?, state_name = ?, state_timestamp = ?, updated = ?,
252
-
\\ start_time = ?, end_time = ?, run_count = ?, total_run_time = ?,
253
-
\\ expected_start_time = COALESCE(?, expected_start_time),
254
-
\\ next_scheduled_start_time = ?
255
-
\\WHERE id = ?
256
-
, .{
257
-
state_id,
258
-
state_type,
259
-
state_name,
260
-
timestamp,
261
-
timestamp,
262
-
start_time,
263
-
end_time,
264
-
run_count,
265
-
total_run_time,
266
-
expected_start_time,
267
-
next_scheduled_start_time,
268
-
run_id,
269
-
}) catch |err| {
270
-
log.err("database", "update flow_run error: {}", .{err});
271
-
return err;
272
-
};
252
+
if (empirical_policy) |policy| {
253
+
txn.exec(
254
+
\\UPDATE flow_run SET state_id = ?, state_type = ?, state_name = ?, state_timestamp = ?, updated = ?,
255
+
\\ start_time = ?, end_time = ?, run_count = ?, total_run_time = ?,
256
+
\\ expected_start_time = COALESCE(?, expected_start_time),
257
+
\\ next_scheduled_start_time = ?,
258
+
\\ empirical_policy = ?
259
+
\\WHERE id = ?
260
+
, .{
261
+
state_id,
262
+
state_type,
263
+
state_name,
264
+
timestamp,
265
+
timestamp,
266
+
start_time,
267
+
end_time,
268
+
run_count,
269
+
total_run_time,
270
+
expected_start_time,
271
+
next_scheduled_start_time,
272
+
policy,
273
+
run_id,
274
+
}) catch |err| {
275
+
log.err("database", "update flow_run error: {}", .{err});
276
+
return err;
277
+
};
278
+
} else {
279
+
txn.exec(
280
+
\\UPDATE flow_run SET state_id = ?, state_type = ?, state_name = ?, state_timestamp = ?, updated = ?,
281
+
\\ start_time = ?, end_time = ?, run_count = ?, total_run_time = ?,
282
+
\\ expected_start_time = COALESCE(?, expected_start_time),
283
+
\\ next_scheduled_start_time = ?
284
+
\\WHERE id = ?
285
+
, .{
286
+
state_id,
287
+
state_type,
288
+
state_name,
289
+
timestamp,
290
+
timestamp,
291
+
start_time,
292
+
end_time,
293
+
run_count,
294
+
total_run_time,
295
+
expected_start_time,
296
+
next_scheduled_start_time,
297
+
run_id,
298
+
}) catch |err| {
299
+
log.err("database", "update flow_run error: {}", .{err});
300
+
return err;
301
+
};
302
+
}
273
303
274
304
txn.exec(
275
305
\\INSERT INTO flow_run_state (id, flow_run_id, type, name, timestamp)
+55
-1
src/orchestration/flow_rules.zig
+55
-1
src/orchestration/flow_rules.zig
···
112
112
/// the transition to FAILED is rejected and an AwaitingRetry (SCHEDULED) state
113
113
/// is proposed instead.
114
114
///
115
+
/// policy mutations:
116
+
/// - on retry: sets retry_type="in_process", resuming=false, pause_keys=[]
117
+
/// - on exhaustion: clears retry_type to allow future UI-triggered retries
118
+
///
115
119
/// requires: run_count, retries, retry_delay in RuleContext
116
-
/// output: retry_state_type, retry_state_name, retry_scheduled_time
120
+
/// output: retry_state_type, retry_state_name, retry_scheduled_time, policy flags
117
121
pub const RetryFailedFlows = OrchestrationRule{
118
122
.name = "RetryFailedFlows",
119
123
.from_states = StateTypeSet.init(&.{.RUNNING}),
···
126
130
const max_retries = ctx.retries orelse return;
127
131
128
132
// if run_count > retries, allow transition to FAILED (exhausted retries)
133
+
// clear retry_type to allow for future infrastructure-level retries (e.g. via UI)
129
134
if (ctx.run_count > max_retries) {
135
+
ctx.clearRetryType();
130
136
return;
131
137
}
132
138
···
547
553
// retry should be scheduled for ~now (delay = 0)
548
554
try testing.expect(ctx.retry_scheduled_time != null);
549
555
}
556
+
557
+
test "RetryFailedFlows sets policy mutation flags on retry" {
558
+
const testing = std.testing;
559
+
560
+
var ctx = RuleContext{
561
+
.initial_state = .RUNNING,
562
+
.proposed_state = .FAILED,
563
+
.initial_state_timestamp = "2024-01-19T16:30:00Z",
564
+
.proposed_state_timestamp = "2024-01-19T16:30:01Z",
565
+
.run_id = "test-run-id",
566
+
.run_count = 1,
567
+
.retries = 3,
568
+
.retry_delay = 10,
569
+
};
570
+
571
+
rules.applyPolicy(&CoreFlowPolicy, &ctx);
572
+
573
+
// verify policy mutation flags are set for retry
574
+
try testing.expect(ctx.set_retry_type_in_process);
575
+
try testing.expect(ctx.set_resuming_false);
576
+
try testing.expect(ctx.clear_pause_keys);
577
+
try testing.expect(!ctx.clear_retry_type); // should NOT clear retry_type when retrying
578
+
}
579
+
580
+
test "RetryFailedFlows clears retry_type when retries exhausted" {
581
+
const testing = std.testing;
582
+
583
+
var ctx = RuleContext{
584
+
.initial_state = .RUNNING,
585
+
.proposed_state = .FAILED,
586
+
.initial_state_timestamp = "2024-01-19T16:30:00Z",
587
+
.proposed_state_timestamp = "2024-01-19T16:30:01Z",
588
+
.run_id = "test-run-id",
589
+
.run_count = 3, // already ran 3 times
590
+
.retries = 2, // only 2 retries allowed (exhausted)
591
+
};
592
+
593
+
rules.applyPolicy(&CoreFlowPolicy, &ctx);
594
+
595
+
// transition should be accepted (no more retries)
596
+
try testing.expectEqual(rules.ResponseStatus.ACCEPT, ctx.result.status);
597
+
// clear_retry_type should be set to allow future UI-triggered retries
598
+
try testing.expect(ctx.clear_retry_type);
599
+
// other flags should NOT be set
600
+
try testing.expect(!ctx.set_retry_type_in_process);
601
+
try testing.expect(!ctx.set_resuming_false);
602
+
try testing.expect(!ctx.clear_pause_keys);
603
+
}
+16
src/orchestration/rules.zig
+16
src/orchestration/rules.zig
···
45
45
retry_state_name: ?[]const u8 = null,
46
46
retry_scheduled_time: ?[]const u8 = null,
47
47
48
+
// empirical_policy mutation flags (set by RetryFailedFlows)
49
+
// when set, API handler should update the policy JSON accordingly
50
+
set_retry_type_in_process: bool = false, // set retry_type = "in_process"
51
+
clear_retry_type: bool = false, // set retry_type = null (when retries exhausted)
52
+
clear_pause_keys: bool = false, // set pause_keys = []
53
+
set_resuming_false: bool = false, // set resuming = false
54
+
48
55
/// reject the transition with a reason
49
56
pub fn reject(self: *RuleContext, reason: []const u8) void {
50
57
self.result.status = .REJECT;
···
75
82
self.retry_state_type = .SCHEDULED;
76
83
self.retry_state_name = "AwaitingRetry";
77
84
self.retry_scheduled_time = scheduled_time;
85
+
// set policy mutation flags for retry
86
+
self.set_retry_type_in_process = true;
87
+
self.set_resuming_false = true;
88
+
self.clear_pause_keys = true;
78
89
log.debug("orchestration", "rule scheduled retry: {s} at {s}", .{ reason, scheduled_time });
90
+
}
91
+
92
+
/// signal that retries are exhausted and retry_type should be cleared
93
+
pub fn clearRetryType(self: *RuleContext) void {
94
+
self.clear_retry_type = true;
79
95
}
80
96
81
97
/// check if transition is still accepted (not yet rejected/waited/aborted)