From 58020afc75f0e8cdaefcf9a02e57da04baa9ebf3 Mon Sep 17 00:00:00 2001 From: Guy Date: Thu, 20 Aug 2026 19:48:23 +0300 Subject: [PATCH] =?UTF-8?q?feat(api):=20failRun=20dead=5Fletter=20flag=20?= =?UTF-8?q?=E2=80=94=20client-forced=20dead-letter?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /runs/{id}/fail accepts optional "dead_letter": true → skip max_attempts check, take dead-letter branch (reason client_requested, next_eligible=impossible, claim SQL excludes it). Use case: deterministic errors (config/definition problems) can never succeed on retry. Without this, a task with max_attempts NULL retries forever, or burns its full attempt budget on an error that is not transient. Also wires module tests into zig build test (previously 0 collected): main.zig test block imports store/api/domain/config; fixes latent compile errors (compat Dir wrap, ObjectMap.deinit(alloc)) and arena leaks in 3 auth tests. --- src/api.zig | 15 ++++++--- src/config.zig | 10 +++--- src/main.zig | 8 +++++ src/store.zig | 84 ++++++++++++++++++++++++++++++++++++++++++-------- 4 files changed, 95 insertions(+), 22 deletions(-) diff --git a/src/api.zig b/src/api.zig index 161011e..2872b31 100644 --- a/src/api.zig +++ b/src/api.zig @@ -985,6 +985,7 @@ fn handleFail(ctx: *Context, run_id: []const u8, body: []const u8, raw_request: var parsed = std.json.parseFromSlice(struct { @"error": []const u8, usage: ?std.json.Value = null, + dead_letter: bool = false, }, ctx.allocator, body, .{ .ignore_unknown_fields = true }) catch { return respondError(ctx.allocator, 400, "invalid_json", "Invalid JSON body"); }; @@ -992,7 +993,7 @@ fn handleFail(ctx: *Context, run_id: []const u8, body: []const u8, raw_request: const req = parsed.value; const usage_json = if (req.usage) |u| (jsonStringify(ctx.allocator, u) catch null) else null; - ctx.store.failRun(run_id, req.@"error", usage_json) catch return serverError(ctx.allocator); + ctx.store.failRun(run_id, req.@"error", usage_json, req.dead_letter) catch return serverError(ctx.allocator); return .{ .status = "200 OK", .body = "{\"status\":\"failed\"}" }; } @@ -1692,12 +1693,14 @@ fn serverError(allocator: std.mem.Allocator) HttpResponse { } test "auth allows health without API token" { + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena.deinit(); var store = try Store.init(std.testing.allocator, ":memory:"); defer store.deinit(); var ctx = Context{ .store = &store, - .allocator = std.testing.allocator, + .allocator = arena.allocator(), .required_api_token = "secret", }; @@ -1706,12 +1709,14 @@ test "auth allows health without API token" { } test "auth rejects protected endpoint without API token" { + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena.deinit(); var store = try Store.init(std.testing.allocator, ":memory:"); defer store.deinit(); var ctx = Context{ .store = &store, - .allocator = std.testing.allocator, + .allocator = arena.allocator(), .required_api_token = "secret", }; @@ -1720,12 +1725,14 @@ test "auth rejects protected endpoint without API token" { } test "auth accepts admin token for protected endpoint" { + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena.deinit(); var store = try Store.init(std.testing.allocator, ":memory:"); defer store.deinit(); var ctx = Context{ .store = &store, - .allocator = std.testing.allocator, + .allocator = arena.allocator(), .required_api_token = "secret", }; diff --git a/src/config.zig b/src/config.zig index e124be4..0c4d2bb 100644 --- a/src/config.zig +++ b/src/config.zig @@ -81,7 +81,7 @@ test "loadFromFile reads config values" { var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); - try tmp.dir.writeFile(.{ + try std_compat.fs.Dir.wrap(tmp.dir).writeFile(.{ .sub_path = "config.json", .data = \\{ @@ -92,7 +92,7 @@ test "loadFromFile reads config values" { , }); - const cfg_path = try tmp.dir.realpathAlloc(std.testing.allocator, "config.json"); + const cfg_path = try std_compat.fs.Dir.wrap(tmp.dir).realpathAlloc(std.testing.allocator, "config.json"); defer std.testing.allocator.free(cfg_path); var arena = std.heap.ArenaAllocator.init(std.testing.allocator); @@ -108,8 +108,8 @@ test "resolveRelativePaths anchors db to config directory" { var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); - try tmp.dir.makePath("configs"); - try tmp.dir.writeFile(.{ + try std_compat.fs.Dir.wrap(tmp.dir).makePath("configs"); + try std_compat.fs.Dir.wrap(tmp.dir).writeFile(.{ .sub_path = "configs/config.json", .data = \\{ @@ -118,7 +118,7 @@ test "resolveRelativePaths anchors db to config directory" { , }); - const cfg_path = try tmp.dir.realpathAlloc(std.testing.allocator, "configs/config.json"); + const cfg_path = try std_compat.fs.Dir.wrap(tmp.dir).realpathAlloc(std.testing.allocator, "configs/config.json"); defer std.testing.allocator.free(cfg_path); var arena = std.heap.ArenaAllocator.init(std.testing.allocator); diff --git a/src/main.zig b/src/main.zig index 50a84f6..1adcffe 100644 --- a/src/main.zig +++ b/src/main.zig @@ -233,3 +233,11 @@ fn readHttpRequest(allocator: std.mem.Allocator, stream: *std.Io.net.Stream, max return try allocator.dupe(u8, buffer.items[0..required]); } + +// Pull module test declarations into the test build (zig build test). +test { + _ = @import("store.zig"); + _ = @import("api.zig"); + _ = @import("domain.zig"); + _ = @import("config.zig"); +} diff --git a/src/store.zig b/src/store.zig index 14bbe7d..d60c71b 100644 --- a/src/store.zig +++ b/src/store.zig @@ -1312,7 +1312,7 @@ pub const Store = struct { // ===== Fail ===== - pub fn failRun(self: *Self, run_id: []const u8, error_text: []const u8, usage_json: ?[]const u8) !void { + pub fn failRun(self: *Self, run_id: []const u8, error_text: []const u8, usage_json: ?[]const u8, force_dead_letter: bool) !void { try self.execSimple("BEGIN IMMEDIATE;"); errdefer self.execSimple("ROLLBACK;") catch {}; @@ -1368,8 +1368,9 @@ pub const Store = struct { fail_count = c.sqlite3_column_int64(cnt_stmt, 0); } - const exhausted = if (max_attempts) |limit| fail_count >= limit else false; + const exhausted = force_dead_letter or (if (max_attempts) |limit| fail_count >= limit else false); if (exhausted) { + const dead_letter_reason: []const u8 = if (force_dead_letter) "client_requested" else "max_attempts_exceeded"; var dead_stage_to_use: ?[]const u8 = null; if (dead_letter_stage) |candidate| { const pip_stmt = try self.prepare("SELECT definition_json FROM pipelines WHERE id = ?;"); @@ -1390,24 +1391,26 @@ pub const Store = struct { const impossible_retry_ts: i64 = 9_223_372_036_854_775_000; if (dead_stage_to_use) |stage| { - const upd = try self.prepare("UPDATE tasks SET stage = ?, task_version = task_version + 1, dead_letter_reason = 'max_attempts_exceeded', next_eligible_at_ms = ?, updated_at_ms = ? WHERE id = ?;"); + const upd = try self.prepare("UPDATE tasks SET stage = ?, task_version = task_version + 1, dead_letter_reason = ?, next_eligible_at_ms = ?, updated_at_ms = ? WHERE id = ?;"); defer _ = c.sqlite3_finalize(upd); self.bindText(upd, 1, stage); - _ = c.sqlite3_bind_int64(upd, 2, impossible_retry_ts); - _ = c.sqlite3_bind_int64(upd, 3, now_ms); - self.bindText(upd, 4, task_id); + self.bindText(upd, 2, dead_letter_reason); + _ = c.sqlite3_bind_int64(upd, 3, impossible_retry_ts); + _ = c.sqlite3_bind_int64(upd, 4, now_ms); + self.bindText(upd, 5, task_id); _ = c.sqlite3_step(upd); } else { - const upd = try self.prepare("UPDATE tasks SET dead_letter_reason = 'max_attempts_exceeded', next_eligible_at_ms = ?, updated_at_ms = ? WHERE id = ?;"); + const upd = try self.prepare("UPDATE tasks SET dead_letter_reason = ?, next_eligible_at_ms = ?, updated_at_ms = ? WHERE id = ?;"); defer _ = c.sqlite3_finalize(upd); - _ = c.sqlite3_bind_int64(upd, 1, impossible_retry_ts); - _ = c.sqlite3_bind_int64(upd, 2, now_ms); - self.bindText(upd, 3, task_id); + self.bindText(upd, 1, dead_letter_reason); + _ = c.sqlite3_bind_int64(upd, 2, impossible_retry_ts); + _ = c.sqlite3_bind_int64(upd, 3, now_ms); + self.bindText(upd, 4, task_id); _ = c.sqlite3_step(upd); } const evt_data = std.json.Stringify.valueAlloc(temp_alloc, .{ - .reason = "max_attempts_exceeded", + .reason = dead_letter_reason, .failed_attempts = fail_count, .from_stage = current_stage, .dead_letter_stage = dead_stage_to_use, @@ -2422,7 +2425,7 @@ test "claim respects per-state concurrency limits" { // Set per-state concurrency limit of 2 for "review" var concurrency_map: std.json.ObjectMap = .empty; - defer concurrency_map.deinit(); + defer concurrency_map.deinit(alloc); try concurrency_map.put(alloc, "review", .{ .integer = 2 }); const per_state: std.json.Value = .{ .object = concurrency_map }; @@ -2520,7 +2523,7 @@ test "fail run with retry policy" { // First claim and fail const c1 = (try store.claimTask("agent-1", "worker", 300_000, null)).?; defer store.freeClaimResult(c1); - try store.failRun(c1.run.id, "error 1", null); + try store.failRun(c1.run.id, "error 1", null, false); // Task should have next_eligible_at_ms set (retry delay) const task_after_1 = (try store.getTask(task_id)).?; @@ -2528,3 +2531,58 @@ test "fail run with retry policy" { try std.testing.expect(task_after_1.next_eligible_at_ms > 0); try std.testing.expect(task_after_1.dead_letter_reason == null); } + +test "forced dead-letter bypasses missing max_attempts" { + const alloc = std.testing.allocator; + var store = try Store.init(alloc, ":memory:"); + defer store.deinit(); + + const pipeline_def = + \\{"initial":"process","states":{"process":{"agent_role":"worker"},"done":{"terminal":true}},"transitions":[{"from":"process","to":"done","trigger":"complete"}]} + ; + + const pipeline_id = try store.createPipeline("forced-dead-letter-test", pipeline_def); + defer store.freeOwnedString(pipeline_id); + + // max_attempts NULL — normal path would retry forever + const task_id = try store.createTask(pipeline_id, "Config Error Task", "desc", 0, "{}", null, 0, null); + defer store.freeOwnedString(task_id); + + const c1 = (try store.claimTask("agent-1", "worker", 300_000, null)).?; + defer store.freeClaimResult(c1); + try store.failRun(c1.run.id, "no workflow for pipeline", null, true); + + const task_after = (try store.getTask(task_id)).?; + defer store.freeTaskRow(task_after); + try std.testing.expectEqualStrings("client_requested", task_after.dead_letter_reason.?); + try std.testing.expect(task_after.next_eligible_at_ms > 9_000_000_000_000_000_000); + + // Dead-lettered task is not claimable + const reclaim = try store.claimTask("agent-1", "worker", 300_000, null); + try std.testing.expect(reclaim == null); +} + +test "unflagged failRun with max_attempts NULL schedules retry" { + const alloc = std.testing.allocator; + var store = try Store.init(alloc, ":memory:"); + defer store.deinit(); + + const pipeline_def = + \\{"initial":"process","states":{"process":{"agent_role":"worker"},"done":{"terminal":true}},"transitions":[{"from":"process","to":"done","trigger":"complete"}]} + ; + + const pipeline_id = try store.createPipeline("unflagged-retry-test", pipeline_def); + defer store.freeOwnedString(pipeline_id); + + const task_id = try store.createTask(pipeline_id, "Retry Task", "desc", 0, "{}", null, 500, null); + defer store.freeOwnedString(task_id); + + const c1 = (try store.claimTask("agent-1", "worker", 300_000, null)).?; + defer store.freeClaimResult(c1); + try store.failRun(c1.run.id, "error 1", null, false); + + const task_after = (try store.getTask(task_id)).?; + defer store.freeTaskRow(task_after); + try std.testing.expect(task_after.dead_letter_reason == null); + try std.testing.expect(task_after.next_eligible_at_ms > 0); +}