zig port of github.com/chrisguidry/docket
memory backend via burner-redis + integration tests
Wires `tangled.org/zzstoatzz.io/burner-redis` (zig 0.16 port of
PrefectHQ/burner-redis, in-process Redis-compatible engine) as
docket's second backend.
src/backend/memory.zig
- MemoryBackend owns a burner.Store + burner.scripting.LuaEngine.
- scheduleTask / promoteDueTasks / ackMessage EVAL the SAME three
Lua scripts the redis backend uses — single source of truth at
src/scripts/*.lua, ports cleanly across backends.
- ensureConsumerGroup + consumeStream skip Lua and call straight
into the store (matches the redis backend's xgroupCreate /
xreadgroup path, which also doesn't go through EVAL).
- read_arena holds the bytes of the most recent consumeStream
result; reset on each call so previous slices stop being valid,
matching the redis client's read-buffer behaviour.
src/backend.zig
- Backend union learns the .memory variant.
- connect() now resolves memory:// URLs to MemoryBackend instances.
tests/integration.zig
- drives Docket end-to-end against both backends through the
perpetual schedule → promote → consume → ack → reschedule cycle.
- memory backend tests run unconditionally (no external services).
- redis backend test gated on DOCKET_TEST_REDIS_URL so CI / local
runs can opt in.
- new `zig build test-integration` step.
Smoke test (manual):
DOCKET_URL=memory://docket-example ./zig-out/bin/perpetual
→ 4 perpetual ticks observed in 10s (50ms interval), clean exit.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
+425
-4
+27
-1
build.zig
+27
-1
build.zig
···
12
12
.target = target,
13
13
.optimize = optimize,
14
14
});
15
+
const burner_dep = b.dependency("burner_redis", .{
16
+
.target = target,
17
+
.optimize = optimize,
18
+
});
15
19
16
20
const imports: []const std.Build.Module.Import = &.{
17
21
.{ .name = "redis", .module = redis_dep.module("redis") },
18
22
.{ .name = "uuid", .module = uuid_dep.module("uuid") },
23
+
.{ .name = "burner_redis", .module = burner_dep.module("burner_redis") },
19
24
};
20
25
21
26
// docket library module — exported for downstream consumers.
···
48
53
}),
49
54
});
50
55
const run_unit_tests = b.addRunArtifact(unit_tests);
51
-
const test_step = b.step("test", "run unit tests");
56
+
57
+
// integration tests — drive Docket end-to-end against both backends.
58
+
// memory backend runs unconditionally; redis backend gated on
59
+
// DOCKET_TEST_REDIS_URL (set it to skip-or-run that subset).
60
+
const integration_tests = b.addTest(.{
61
+
.root_module = b.createModule(.{
62
+
.root_source_file = b.path("tests/integration.zig"),
63
+
.target = target,
64
+
.optimize = optimize,
65
+
.imports = &.{
66
+
.{ .name = "docket", .module = docket_module },
67
+
},
68
+
.link_libc = true,
69
+
}),
70
+
});
71
+
const run_integration_tests = b.addRunArtifact(integration_tests);
72
+
73
+
const test_step = b.step("test", "run all tests (unit + integration)");
52
74
test_step.dependOn(&run_unit_tests.step);
75
+
test_step.dependOn(&run_integration_tests.step);
76
+
77
+
const integration_step = b.step("test-integration", "run only the integration tests");
78
+
integration_step.dependOn(&run_integration_tests.step);
53
79
54
80
// example: perpetual — requires a running Redis. `zig build perpetual` to compile.
55
81
const perpetual_exe = b.addExecutable(.{
+4
build.zig.zon
+4
build.zig.zon
···
13
13
.url = "https://codeberg.org/r4gus/uuid-zig/archive/0.5.0.tar.gz",
14
14
.hash = "uuid-0.5.0-oOieIQx-AABtc9U3ihv_bf9pZAYoKp4UMsNFHBr-cn-w",
15
15
},
16
+
.burner_redis = .{
17
+
.url = "https://tangled.org/zzstoatzz.io/burner-redis/archive/5b4ea2c.tar.gz",
18
+
.hash = "burner_redis_zig-0.0.1-JKoFUMpdAQAM63JmLoN-KN2Lw5CZPBy6I8XUAHskGJM6",
19
+
},
16
20
},
17
21
18
22
.paths = .{
+15
-3
src/backend.zig
+15
-3
src/backend.zig
···
14
14
const redis = @import("redis");
15
15
16
16
pub const redis_backend = @import("backend/redis.zig");
17
+
pub const memory_backend = @import("backend/memory.zig");
17
18
pub const RedisBackend = redis_backend.RedisBackend;
19
+
pub const MemoryBackend = memory_backend.MemoryBackend;
18
20
pub const ScheduleArgs = redis_backend.ScheduleArgs;
19
21
pub const PromoteResult = redis_backend.PromoteResult;
20
22
pub const Disposition = redis_backend.Disposition;
21
23
22
-
/// Tagged union over all backend variants.
24
+
/// Tagged union over all backend variants. Both expose the same composite-op
25
+
/// surface; consumers (Docket / Worker) don't know which one they have.
23
26
pub const Backend = union(enum) {
24
27
redis: *RedisBackend,
25
-
// memory: *MemoryBackend, // v0.2 — embedded Redis-compatible in-process server
28
+
memory: *MemoryBackend,
26
29
27
30
pub fn close(self: Backend, io: Io) void {
28
31
switch (self) {
···
73
76
};
74
77
75
78
/// Resolve a URL to a concrete backend.
79
+
///
80
+
/// Supported schemes:
81
+
/// - `redis://host[:port][/db]` — talks to a real Redis (or compatible).
82
+
/// - `memory://<name>` — embeds burner-redis in-process; `<name>`
83
+
/// is informational (key prefix is owned
84
+
/// by the Docket name, not the URL).
76
85
pub fn connect(allocator: Allocator, io: Io, url: []const u8, prefix: []const u8) !Backend {
77
86
if (std.mem.startsWith(u8, url, "redis://") or std.mem.startsWith(u8, url, "unix://")) {
78
87
const rb = try allocator.create(RedisBackend);
···
80
89
return .{ .redis = rb };
81
90
}
82
91
if (std.mem.startsWith(u8, url, "memory://")) {
83
-
return error.MemoryBackendNotYetImplemented;
92
+
const mb = try allocator.create(MemoryBackend);
93
+
mb.* = try MemoryBackend.connect(allocator, io, prefix);
94
+
mb.bindEngine();
95
+
return .{ .memory = mb };
84
96
}
85
97
return error.UnsupportedBackendUrl;
86
98
}
+244
src/backend/memory.zig
+244
src/backend/memory.zig
···
1
+
//! Memory backend — composite ops backed by the embedded burner-redis engine
2
+
//! (`tangled.org/zzstoatzz.io/burner-redis`). The same three Lua scripts that
3
+
//! the Redis backend EVALs remotely run here against an in-process store
4
+
//! through `burner_redis.LuaEngine`.
5
+
//!
6
+
//! What this gives us:
7
+
//! - Self-contained docket adoption — no external Redis required.
8
+
//! - Identical wire semantics: the EVAL scripts are byte-for-byte the same.
9
+
//! - Same `Backend` surface as `RedisBackend`, so `Docket` doesn't care
10
+
//! which backend it has under the hood.
11
+
//!
12
+
//! For ops Redis exposes outside of EVAL (XGROUP CREATE, XREADGROUP), we
13
+
//! call straight into `burner_redis.Store` rather than round-tripping
14
+
//! through Lua — those ops have a stable, narrow shape that's easier to
15
+
//! drive directly.
16
+
17
+
const std = @import("std");
18
+
const Allocator = std.mem.Allocator;
19
+
const Io = std.Io;
20
+
const burner = @import("burner_redis");
21
+
const redis = @import("redis");
22
+
23
+
const redis_backend = @import("redis.zig");
24
+
25
+
pub const ScheduleArgs = redis_backend.ScheduleArgs;
26
+
pub const PromoteResult = redis_backend.PromoteResult;
27
+
pub const Disposition = redis_backend.Disposition;
28
+
29
+
/// Embed the same Lua scripts the Redis backend uses. Single source of
30
+
/// truth lives at src/scripts/*.lua.
31
+
const SCHEDULE_LUA = @embedFile("../scripts/schedule.lua");
32
+
const PROMOTE_DUE_LUA = @embedFile("../scripts/promote_due.lua");
33
+
const ACK_LUA = @embedFile("../scripts/ack.lua");
34
+
35
+
pub const MemoryBackend = struct {
36
+
allocator: Allocator,
37
+
/// Docket logical name — used as the prefix for all keys, matching the
38
+
/// redis backend's behaviour.
39
+
prefix: []const u8,
40
+
store: burner.Store,
41
+
engine: burner.scripting.LuaEngine,
42
+
/// Arena holding the bytes of the most recent consumeStream result.
43
+
/// Reset on each call so previous slices stop being valid, matching the
44
+
/// Redis client's read-buffer behaviour.
45
+
read_arena: std.heap.ArenaAllocator,
46
+
47
+
pub fn connect(allocator: Allocator, io: Io, prefix: []const u8) !MemoryBackend {
48
+
const store = burner.Store.init(allocator, io);
49
+
return .{
50
+
.allocator = allocator,
51
+
.prefix = prefix,
52
+
.store = store,
53
+
.engine = burner.scripting.LuaEngine.init(allocator, undefined),
54
+
.read_arena = std.heap.ArenaAllocator.init(allocator),
55
+
};
56
+
}
57
+
58
+
/// Engine has to hold a `*Store` pointer; we patch it here once the
59
+
/// embedded store has a stable address. Called by `connect_alloc` (the
60
+
/// `*MemoryBackend` factory in backend.zig).
61
+
pub fn bindEngine(self: *MemoryBackend) void {
62
+
self.engine = burner.scripting.LuaEngine.init(self.allocator, &self.store);
63
+
}
64
+
65
+
pub fn close(self: *MemoryBackend, _: Io) void {
66
+
self.read_arena.deinit();
67
+
self.store.deinit();
68
+
self.allocator.destroy(self);
69
+
}
70
+
71
+
// ------------------------------------------------------------------------
72
+
// Composite ops — match RedisBackend's surface byte-for-byte.
73
+
// ------------------------------------------------------------------------
74
+
75
+
pub fn scheduleTask(self: *MemoryBackend, args: ScheduleArgs) !Disposition {
76
+
var arena = std.heap.ArenaAllocator.init(self.allocator);
77
+
defer arena.deinit();
78
+
const a = arena.allocator();
79
+
80
+
const stream_key = try keyJoin(a, &.{ self.prefix, "stream" });
81
+
const parked_key = try keyJoin(a, &.{ self.prefix, "parked", args.task_key });
82
+
const queue_key = try keyJoin(a, &.{ self.prefix, "queue" });
83
+
const runs_key = try keyJoin(a, &.{ self.prefix, "runs", args.task_key });
84
+
85
+
const when_str = try std.fmt.allocPrint(a, "{d}", .{args.when_micros});
86
+
const is_immediate_str: []const u8 = if (args.is_immediate) "1" else "0";
87
+
88
+
const keys = [_][]const u8{ stream_key, parked_key, queue_key, runs_key };
89
+
90
+
var argv: std.ArrayList([]const u8) = .empty;
91
+
defer argv.deinit(a);
92
+
try argv.appendSlice(a, &.{ args.task_key, when_str, is_immediate_str });
93
+
try argv.appendSlice(a, args.fields);
94
+
95
+
const reply = try self.engine.eval(a, SCHEDULE_LUA, &keys, argv.items);
96
+
// Reply is either "SCHEDULED" (success) or an error.
97
+
return switch (reply) {
98
+
.bulk_string => |s| if (std.mem.eql(u8, s, "SCHEDULED"))
99
+
Disposition.scheduled
100
+
else
101
+
error.UnexpectedScheduleReply,
102
+
.status => |s| if (std.mem.eql(u8, s, "SCHEDULED"))
103
+
Disposition.scheduled
104
+
else
105
+
error.UnexpectedScheduleReply,
106
+
.err => |msg| {
107
+
std.log.warn("memory backend schedule error: {s}", .{msg});
108
+
return error.RedisCommandFailed;
109
+
},
110
+
else => error.UnexpectedScheduleReply,
111
+
};
112
+
}
113
+
114
+
pub fn promoteDueTasks(self: *MemoryBackend, now_micros: i64) !PromoteResult {
115
+
var arena = std.heap.ArenaAllocator.init(self.allocator);
116
+
defer arena.deinit();
117
+
const a = arena.allocator();
118
+
119
+
const queue_key = try keyJoin(a, &.{ self.prefix, "queue" });
120
+
const stream_key = try keyJoin(a, &.{ self.prefix, "stream" });
121
+
const now_str = try std.fmt.allocPrint(a, "{d}", .{now_micros});
122
+
123
+
const keys = [_][]const u8{ queue_key, stream_key };
124
+
const argv = [_][]const u8{ now_str, self.prefix };
125
+
126
+
const reply = try self.engine.eval(a, PROMOTE_DUE_LUA, &keys, &argv);
127
+
return switch (reply) {
128
+
.array => |arr| blk: {
129
+
if (arr.len != 2) return error.UnexpectedPromoteReply;
130
+
const total = switch (arr[0]) {
131
+
.integer => |i| i,
132
+
else => return error.UnexpectedPromoteReply,
133
+
};
134
+
const promoted = switch (arr[1]) {
135
+
.integer => |i| i,
136
+
else => return error.UnexpectedPromoteReply,
137
+
};
138
+
break :blk .{ .total_in_queue = total, .promoted = promoted };
139
+
},
140
+
.err => |msg| {
141
+
std.log.warn("promote_due lua error: {s}", .{msg});
142
+
return error.RedisCommandFailed;
143
+
},
144
+
else => error.UnexpectedPromoteReply,
145
+
};
146
+
}
147
+
148
+
pub fn ackMessage(
149
+
self: *MemoryBackend,
150
+
group: []const u8,
151
+
msg_id: []const u8,
152
+
task_key: []const u8,
153
+
terminal_state: []const u8,
154
+
) !void {
155
+
var arena = std.heap.ArenaAllocator.init(self.allocator);
156
+
defer arena.deinit();
157
+
const a = arena.allocator();
158
+
159
+
const stream_key = try keyJoin(a, &.{ self.prefix, "stream" });
160
+
const runs_key = try keyJoin(a, &.{ self.prefix, "runs", task_key });
161
+
162
+
const keys = [_][]const u8{ stream_key, runs_key };
163
+
const argv = [_][]const u8{ group, msg_id, terminal_state };
164
+
165
+
const reply = try self.engine.eval(a, ACK_LUA, &keys, &argv);
166
+
switch (reply) {
167
+
.bulk_string => |s| if (!std.mem.eql(u8, s, "OK")) return error.UnexpectedAckReply,
168
+
.status => |s| if (!std.mem.eql(u8, s, "OK")) return error.UnexpectedAckReply,
169
+
.err => return error.RedisCommandFailed,
170
+
else => return error.UnexpectedAckReply,
171
+
}
172
+
}
173
+
174
+
pub fn ensureConsumerGroup(self: *MemoryBackend, group: []const u8) !void {
175
+
var arena = std.heap.ArenaAllocator.init(self.allocator);
176
+
defer arena.deinit();
177
+
const a = arena.allocator();
178
+
179
+
const stream_key = try keyJoin(a, &.{ self.prefix, "stream" });
180
+
self.store.xgroupCreate(stream_key, group, .{
181
+
.id = burner.store.StreamId.zero(),
182
+
.mkstream = true,
183
+
}) catch |err| switch (err) {
184
+
burner.store.StoreError.BusyGroup => return, // idempotent
185
+
else => return err,
186
+
};
187
+
}
188
+
189
+
/// XREADGROUP one entry. Returns null if no new messages. The returned
190
+
/// StreamEntry's slices live in `self.read_arena`, valid until the next
191
+
/// call to consumeStream. This matches the redis client's semantics
192
+
/// (StreamEntry references the client's read buffer).
193
+
pub fn consumeStream(
194
+
self: *MemoryBackend,
195
+
group: []const u8,
196
+
consumer: []const u8,
197
+
block_ms: u32,
198
+
) !?redis.StreamEntry {
199
+
_ = block_ms; // in-process: no blocking — single-poll semantics.
200
+
201
+
// Reset the read arena so the previous call's slices stop being valid.
202
+
self.read_arena.deinit();
203
+
self.read_arena = std.heap.ArenaAllocator.init(self.allocator);
204
+
const a = self.read_arena.allocator();
205
+
206
+
const stream_key = try keyJoin(a, &.{ self.prefix, "stream" });
207
+
208
+
const delivered = try self.store.xreadgroupOne(a, stream_key, group, consumer, .new_messages, 1);
209
+
if (delivered.len == 0) return null;
210
+
const de = delivered[0];
211
+
212
+
// Format id as "ms-seq" and copy fields into arena-owned StreamField
213
+
// slices so the StreamEntry matches the redis client's shape.
214
+
const id_str = try std.fmt.allocPrint(a, "{d}-{d}", .{ de.id.ms, de.id.seq });
215
+
216
+
var fields = try a.alloc(redis.StreamField, de.fields.len);
217
+
for (de.fields, 0..) |p, i| {
218
+
fields[i] = .{ .field = p.field, .value = p.value };
219
+
}
220
+
221
+
return .{ .id = id_str, .fields = fields };
222
+
}
223
+
224
+
};
225
+
226
+
/// Build a Redis key from colon-joined parts. Caller owns the result.
227
+
fn keyJoin(alloc: Allocator, parts: []const []const u8) ![]const u8 {
228
+
var total: usize = 0;
229
+
for (parts, 0..) |p, i| {
230
+
total += p.len;
231
+
if (i + 1 < parts.len) total += 1;
232
+
}
233
+
const out = try alloc.alloc(u8, total);
234
+
var pos: usize = 0;
235
+
for (parts, 0..) |p, i| {
236
+
@memcpy(out[pos..][0..p.len], p);
237
+
pos += p.len;
238
+
if (i + 1 < parts.len) {
239
+
out[pos] = ':';
240
+
pos += 1;
241
+
}
242
+
}
243
+
return out;
244
+
}
+135
tests/integration.zig
+135
tests/integration.zig
···
1
+
//! docket integration tests — drive the full schedule → promote → consume →
2
+
//! ack → reschedule cycle through `Docket`, against both backends.
3
+
//!
4
+
//! - memory backend runs unconditionally (no external services required).
5
+
//! - redis backend runs when `DOCKET_TEST_REDIS_URL` is set (e.g. via
6
+
//! `docker compose up redis` + `DOCKET_TEST_REDIS_URL=redis://localhost:6379/0`).
7
+
//!
8
+
//! `zig build test` runs everything that doesn't need Redis. To run the
9
+
//! Redis subset, set the env var and re-run.
10
+
11
+
const std = @import("std");
12
+
const Io = std.Io;
13
+
const docket = @import("docket");
14
+
15
+
// ----------------------------------------------------------------------------
16
+
// shared task body — the smallest perpetual that exercises every path.
17
+
// ----------------------------------------------------------------------------
18
+
19
+
const TickParams = struct { label: []const u8 };
20
+
21
+
const Sink = struct {
22
+
var count: u32 = 0;
23
+
var last_label: [64]u8 = undefined;
24
+
var last_label_len: usize = 0;
25
+
26
+
fn reset() void {
27
+
count = 0;
28
+
last_label_len = 0;
29
+
}
30
+
};
31
+
32
+
fn onTick(_: Io, _: *docket.TaskContext, p: TickParams) !void {
33
+
Sink.count += 1;
34
+
const n = @min(p.label.len, Sink.last_label.len);
35
+
@memcpy(Sink.last_label[0..n], p.label[0..n]);
36
+
Sink.last_label_len = n;
37
+
}
38
+
39
+
// ----------------------------------------------------------------------------
40
+
// shared workout — registers + seeds + drives the worker for `ticks` rounds.
41
+
// ----------------------------------------------------------------------------
42
+
43
+
const Workout = struct {
44
+
fn run(allocator: std.mem.Allocator, io: Io, url: []const u8, name: []const u8) !u32 {
45
+
Sink.reset();
46
+
47
+
var d = try docket.Docket.init(allocator, io, .{ .name = name, .url = url });
48
+
defer d.deinit();
49
+
50
+
try d.register("tick", onTick, TickParams, .{
51
+
.perpetual = .{ .every_micros = 50 * std.time.us_per_ms },
52
+
});
53
+
54
+
try d.backend.ensureConsumerGroup(docket.WORKER_GROUP);
55
+
56
+
const handle = try d.add("tick", TickParams{ .label = "hi" }, .{});
57
+
defer handle.deinit(allocator);
58
+
59
+
var w = try docket.Worker.init(allocator, io, &d, .{
60
+
.name = "worker-1",
61
+
.poll_block_ms = 50,
62
+
.scheduler_interval_micros = 20 * std.time.us_per_ms,
63
+
});
64
+
defer w.deinit();
65
+
66
+
// Drive ~1s of worker ticks; the perpetual interval is 50ms, so we
67
+
// expect on the order of 15-20 ticks. We assert >= 5 to keep the
68
+
// test stable under CI jitter while still proving the loop made
69
+
// progress past the seed.
70
+
const start_ns = Io.Timestamp.now(io, .real).nanoseconds;
71
+
const stop_ns: i128 = start_ns + 1 * std.time.ns_per_s;
72
+
while (Io.Timestamp.now(io, .real).nanoseconds < stop_ns) {
73
+
try w.tick();
74
+
}
75
+
76
+
return Sink.count;
77
+
}
78
+
};
79
+
80
+
// ----------------------------------------------------------------------------
81
+
// memory backend
82
+
// ----------------------------------------------------------------------------
83
+
84
+
test "memory backend: perpetual full cycle" {
85
+
var threaded = Io.Threaded.init(std.testing.allocator, .{});
86
+
defer threaded.deinit();
87
+
88
+
const observed = try Workout.run(std.testing.allocator, threaded.io(), "memory://docket-tests", "docket-tests");
89
+
try std.testing.expect(observed >= 5);
90
+
}
91
+
92
+
test "memory backend: schedule + consume single ack" {
93
+
var threaded = Io.Threaded.init(std.testing.allocator, .{});
94
+
defer threaded.deinit();
95
+
96
+
var d = try docket.Docket.init(std.testing.allocator, threaded.io(), .{
97
+
.name = "single",
98
+
.url = "memory://single",
99
+
});
100
+
defer d.deinit();
101
+
102
+
try d.register("tick", onTick, TickParams, .{});
103
+
try d.backend.ensureConsumerGroup(docket.WORKER_GROUP);
104
+
105
+
Sink.reset();
106
+
const handle = try d.add("tick", TickParams{ .label = "once" }, .{});
107
+
defer handle.deinit(std.testing.allocator);
108
+
109
+
var w = try docket.Worker.init(std.testing.allocator, threaded.io(), &d, .{
110
+
.name = "w",
111
+
.poll_block_ms = 1,
112
+
.scheduler_interval_micros = 1,
113
+
});
114
+
defer w.deinit();
115
+
116
+
// One worker.tick() should pick up the seeded message and run the task.
117
+
try w.tick();
118
+
try std.testing.expectEqual(@as(u32, 1), Sink.count);
119
+
try std.testing.expectEqualStrings("once", Sink.last_label[0..Sink.last_label_len]);
120
+
}
121
+
122
+
// ----------------------------------------------------------------------------
123
+
// redis backend (gated by DOCKET_TEST_REDIS_URL)
124
+
// ----------------------------------------------------------------------------
125
+
126
+
test "redis backend: perpetual full cycle (skipped unless DOCKET_TEST_REDIS_URL set)" {
127
+
const url = std.c.getenv("DOCKET_TEST_REDIS_URL") orelse return error.SkipZigTest;
128
+
const url_slice = std.mem.span(url);
129
+
130
+
var threaded = Io.Threaded.init(std.testing.allocator, .{});
131
+
defer threaded.deinit();
132
+
133
+
const observed = try Workout.run(std.testing.allocator, threaded.io(), url_slice, "docket-redis-tests");
134
+
try std.testing.expect(observed >= 5);
135
+
}