Skip to content

feat(api): failRun dead_letter flag — client-forced dead-letter #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
addadi wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(api): failRun dead_letter flag — client-forced dead-letter #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 11 additions & 4 deletions src/api.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -985,14 +985,15 @@ 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");
};
defer parsed.deinit();
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\"}" };
}
Expand Down Expand Up @@ -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",
};

Expand All @@ -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",
};

Expand All @@ -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",
};

Expand Down
10 changes: 5 additions & 5 deletions src/config.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 =
\\{
Expand All @@ -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);
Expand All @@ -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 =
\\{
Expand All @@ -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);
Expand Down
8 changes: 8 additions & 0 deletions src/main.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");
}
84 changes: 71 additions & 13 deletions src/store.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {};

Expand Down Expand Up @@ -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 = ?;");
Expand All @@ -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,
Expand Down Expand Up @@ -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 };

Expand Down Expand Up @@ -2520,11 +2523,66 @@ 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)).?;
defer store.freeTaskRow(task_after_1);
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);
}