zzstoatzz.io
This repository has no description
reads: hydration, filter/count endpoints, secondary indexes + benchmark
the appview grows up: membership sets per (kind, state / flow / parent)
in burner-redis — the same shape as the cloud prototype's redis run-
state indices — back POST /flow_runs/filter, /count, /task_runs/filter
and /flows/filter. boot now rebuilds the whole appview by replaying the
repo (entities, then the event log in TID order), so reads survive
restarts: the delete-the-cache-replay-the-truth property is exercised
on every boot.
docs/access-patterns.md grounds the design against guidry's proofs in
prefectlabs/object-store-orchestration (latency bands, the delta-size
lever, the disproven many-small-objects pattern) and lays out the
three-tier aggregation story (maintained counters / bounded scans /
CAR-export analytics).
scripts/bench-reads runs the same seeded workload against any prefect
REST target. results (docs/bench-results.md): membership reads 4-11x
faster than python on sqlite or postgres, bulk lists tied at n=300,
202/s writes while signing a commit and rewriting the CAR every
mutation. all four client read paths (point + filter) now pass against
a restarted server.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
author
zzstoatzz
co-author
Claude Fable 5
date
(Jun 12, 2026, 2:54 AM -0500)
commit
41fadeda
41fadeda27c2f222ec487be9a065455fa3286aa9
parent
387e639e
387e639e28b26680c7ff965f0c8394f18a3da6c9
+660
-12
-3
README.md
-3
README.md
···
65
65
## honest limitations (v0)
66
66
67
67
- **no orchestration rules** — every proposed state is ACCEPTed
68
-
- **appview hydration on boot is not implemented yet** — after a restart,
69
-
records are intact in the repo but `/api` reads of prior entities 404
70
-
until replay lands (the boot log warns about this)
71
68
- **events websocket is stubbed** — prefect 3.x local execution reports task
72
69
runs via events, so task-run records only appear when the client posts
73
70
them through the REST path
+100
docs/access-patterns.md
+100
docs/access-patterns.md
···
1
+
# access patterns & aggregations
2
+
3
+
how reads work over an atproto-repo-canonical store, calibrated against the
4
+
cloud prototype (`prefectlabs/object-store-orchestration`, guidry's proofs/
5
+
directory) so the two designs can be compared honestly.
6
+
7
+
## what the cloud prototype proved (and what we borrow)
8
+
9
+
the prototype's GKE simulation (p95/whale workspace, ~440 writes/s, 30 query
10
+
patterns) sorted every read into three latency bands:
11
+
12
+
| band | what answers it | examples (their p50) |
13
+
|---|---|---|
14
+
| **sub-30 ms** | redis directly, or pod-local index cache | `active_by_state` 14ms, deployment reads 5ms |
15
+
| **100–500 ms** | iceberg hour-partition scans through a byte-range cache | `completed_1h` 112ms → `completed_24h` 279ms |
16
+
| **500–900 ms** | delta-merge dominated (hot rows not yet flushed) | `task_runs_24h` 869ms, `by_tag_24h` 918ms |
17
+
18
+
three of their findings transfer directly:
19
+
20
+
1. **the live-state index is a membership structure, not a table.**
21
+
their `active_by_state` is a redis ZSET; ours is a burner-redis set per
22
+
`(kind, state)`. same shape, same O(members) cost, same answer to "what
23
+
is running right now."
24
+
2. **the latency knob is unmerged-delta size, not file count.** their
25
+
`scan_redis_deltas` finding (5-min flush → 830ms; 30-s flush → ~460
26
+
deltas and falling). our analog is commit cadence: everything in the
27
+
appview is always "flushed" (synchronous dual-write), so we sit at the
28
+
degenerate happy case — at the cost of write latency, which is the
29
+
trade to benchmark.
30
+
3. **many small objects is death.** their disproven `parquet-indexes`
31
+
proof: 500 individual delta files = 52–95× degradation. our analog
32
+
decision: one CAR, batched commits — never a file per record. (the
33
+
full-CAR-rewrite-per-commit we do today is the *opposite* extreme and
34
+
fails at the other end; block-level append is the meeting point.)
35
+
36
+
their explicitly open item: **"we haven't run the same queries against the
37
+
production Postgres to compare head-to-head."** that's the gap our
38
+
benchmark fills at small scale (see `scripts/bench-reads`).
39
+
40
+
## the read surface, mapped
41
+
42
+
| pattern (their name) | their path | our path |
43
+
|---|---|---|
44
+
| `run_point_lookup` | redis hash / bucket JSON | appview row get (`flow_run:<id>`) |
45
+
| `active_by_state` | redis ZSET slice | `s:flow_run:state:RUNNING` set → rows |
46
+
| `terminal_completed_*` | iceberg partition scan + delta merge | `s:flow_run:state:COMPLETED` set → rows, ISO-string sort |
47
+
| `paginated_*` | OFFSET over iceberg scan | sort + slice (same O(n), honest about it) |
48
+
| `by_flow`, `by_deployment` | IndexTable predicate pushdown | `s:flow_run:flow:<id>` membership set |
49
+
| `deployment_list` | IndexTable (parquet base + redis delta) | `s:flow:all` → rows, name sort |
50
+
| `count_by_state` | ZSET cardinality | `scard` — O(1) |
51
+
| time-window scans (`completed_1h`) | hour-partition pruning | **TID-range scans (designed, not built):** event rkeys are TIDs, and the MST is ordered — a key-range walk from `TID(t0)` to `TID(t1)` is partition pruning by construction |
52
+
53
+
sorting note: ISO-8601 timestamps sort correctly as strings, so the appview
54
+
never parses dates on the read path.
55
+
56
+
## aggregations: three tiers
57
+
58
+
**tier 1 — maintained aggregates (answers in O(1)).**
59
+
counts by state, by queue, by deployment: set cardinalities updated on every
60
+
transition. this is the cloud prototype's redis counter layer and it covers
61
+
the dashboard's top row. cheap to add: per-day counters
62
+
(`c:flow_run:completed:2026-06-12`) bumped on terminal events.
63
+
64
+
**tier 2 — bounded scans (answers in O(candidates)).**
65
+
everything `/filter` shaped: membership-set intersection → row loads → sort
66
+
→ slice. fine to ~10⁵ rows per workspace in-process. this is where we live
67
+
today and where the benchmark measures us.
68
+
69
+
**tier 3 — analytical queries (answers in O(data), offline-friendly).**
70
+
the cloud answer is iceberg + datafusion, load-bearing for BYOS: "point any
71
+
engine at your bucket." our answer is the same argument with a different
72
+
open format: **the repo CAR is the export**. `getRepo` → decode records →
73
+
parquet → duckdb/datafusion/polars. the pipeline is trivially scriptable
74
+
(`just export-car && duckdb`), needs zero server support, and inherits the
75
+
verifiability story — you can prove the parquet faithfully derives from the
76
+
signed repo. group-bys over months of history don't belong on the
77
+
orchestration hot path in either design.
78
+
79
+
the event log makes tier 3 unusually clean here: `io.prefect.v0.event` is
80
+
already the fact table (occurred, event name, resource id, payload), TID
81
+
rkeys give natural time partitioning, and entity records are the dimension
82
+
tables. it's a star schema by accident.
83
+
84
+
## what we deliberately don't do
85
+
86
+
- no SQL pushdown into the appview — filter shapes are compiled by hand
87
+
(the cloud prototype compiles prefect filter schemas into datafusion
88
+
plans; see their `filter-schema-compilation.md`. if our filter surface
89
+
grows past a handful of shapes, that's the design to copy.)
90
+
- no time-range filters yet (needs TID-range walks or a created-time zset)
91
+
- no cross-entity joins on the hot path — `flow_runs joined to deployments`
92
+
resolves through membership sets or moves to tier 3
93
+
94
+
## benchmark
95
+
96
+
`scripts/bench-reads` seeds an identical workload through the REST API of
97
+
each target, then measures the shared read patterns. honest caveats baked
98
+
in: same process, same machine, sequential requests, demo scale — this
99
+
measures *architecture overhead*, not cloud scale. results in
100
+
`docs/bench-results.md`.
+65
docs/bench-results.md
+65
docs/bench-results.md
···
1
+
# read benchmark — 2026-06-12
2
+
3
+
`scripts/bench-reads`: identical seeded workload (5 flows, 300 flow runs,
4
+
~870 writes via the REST API, seed 1337), then 30 iterations per read
5
+
pattern, sequential, same machine (macOS arm64 M3 Max). prefect-pds is a
6
+
ReleaseFast-equivalent debug build? **no — debug build**; python 3.7.4 via
7
+
`uv run`. pattern names mirror `profile_queries.py` in
8
+
`prefectlabs/object-store-orchestration`.
9
+
10
+
| pattern | prefect-pds p50/p95 | python+sqlite p50/p95 | python+postgres p50/p95 |
11
+
|---|---|---|---|
12
+
| seed write rate | 184–202/s | 160/s | 163/s |
13
+
| point_lookup | **0.2 / 0.3** | 1.7 / 2.3 | 2.6 / 3.3 |
14
+
| active_by_state | **1.0 / 1.2** | 4.4 / 5.5 | 4.7 / 5.2 |
15
+
| completed_50 | 5.5 / 5.9 | 5.8 / 6.5 | 6.7 / 7.2 |
16
+
| paginated (offset 100) | 5.5 / 5.6 | 6.0 / 7.5 | 6.8 / 7.0 |
17
+
| count_completed | **0.2 / 0.3** | 1.3 / 1.8 | 2.1 / 2.7 |
18
+
| flow_list | **0.4 / 0.5** | 3.4 / 4.2 | 4.5 / 4.9 |
19
+
20
+
(milliseconds; python columns from separate runs against fresh stores.)
21
+
22
+
## reading the table
23
+
24
+
- **membership-set reads (point, count, active, list) are 4–11× faster**
25
+
than python against either database — they never leave process memory.
26
+
this is the cloud prototype's "redis answers the orchestration plane"
27
+
band, reproduced in miniature.
28
+
- **bulk list reads are nearly tied** (5.5 vs 5.8–6.8 ms). ours loads all
29
+
candidates then sorts and slices (O(candidates)); sqlite/postgres push
30
+
LIMIT down. at 268 completed runs the costs coincide; at 10⁵ they won't.
31
+
the fix when it matters: a created-time zset so the slice happens before
32
+
the row loads.
33
+
- **write rate: 184–202/s with the repo doing a signed commit + full CAR
34
+
rewrite on every mutation** — and it still edges out python's 160/s.
35
+
signing is not the bottleneck (~0.07ms); the CAR rewrite is, and it's the
36
+
known-naive part (block-level append is the designed fix).
37
+
38
+
## reference points from the cloud prototype (not comparable, but orienting)
39
+
40
+
their GKE simulation runs a p95/whale workspace (~440 writes/s continuous,
41
+
millions of terminal rows) over networked GCS + redis:
42
+
`active_by_state` 14ms, deployment reads 5ms, `completed_24h` 279ms,
43
+
delta-merge-heavy task-run scans 869ms. different scale, different
44
+
infrastructure, different question — theirs is "does the three-plane
45
+
architecture hold at whale scale," ours is "does the atproto-canonical
46
+
architecture cost anything at workspace scale." the answer so far: no —
47
+
the appview makes reads competitive-to-faster, and the canonical layer
48
+
taxes writes ~0 at this scale even with naive persistence.
49
+
50
+
their open item — "we haven't run the same queries against the production
51
+
Postgres" — is partially served here at small scale: python+postgres loses
52
+
to python+sqlite on every pattern at n=300 (network hop dominates), which
53
+
is itself a reminder that these comparisons only mean something within a
54
+
scale class.
55
+
56
+
## caveats
57
+
58
+
- single machine, sequential requests, demo scale (300 runs), one run per
59
+
target. variance not characterized.
60
+
- prefect-pds is a Debug build — its numbers would only improve.
61
+
- python servers measured through uvicorn + pydantic validation; that's
62
+
the product surface, not the database. this benchmarks *architectures
63
+
end-to-end*, not storage engines.
64
+
- no time-window patterns yet (`completed_1h`-style) — needs TID-range
65
+
scans or a time zset on our side.
+168
scripts/bench-reads
+168
scripts/bench-reads
···
1
+
#!/usr/bin/env -S uv run --script --quiet
2
+
# /// script
3
+
# requires-python = ">=3.12"
4
+
# dependencies = ["httpx"]
5
+
# ///
6
+
"""apples-to-apples read benchmark across prefect API implementations.
7
+
8
+
seeds an identical workload through each target's REST API (same seed, same
9
+
order), then measures shared read patterns. pattern names mirror the cloud
10
+
prototype's profile_queries.py so numbers can be lined up side by side.
11
+
12
+
usage:
13
+
./scripts/bench-reads --target pds=http://localhost:4200/api \
14
+
--target python-sqlite=http://localhost:4300/api \
15
+
[--runs 300] [--iterations 30]
16
+
"""
17
+
import argparse
18
+
import json
19
+
import random
20
+
import statistics
21
+
import time
22
+
23
+
import httpx
24
+
25
+
FLOWS = 5
26
+
27
+
28
+
def post_retry(client: httpx.Client, url: str, body: dict) -> httpx.Response:
29
+
"""python prefect sheds 503s under write bursts; retry so seeding
30
+
completes — we are benchmarking reads, not write-overload behavior."""
31
+
for attempt in range(10):
32
+
try:
33
+
r = client.post(url, json=body)
34
+
if r.status_code != 503:
35
+
r.raise_for_status()
36
+
return r
37
+
except (httpx.ReadTimeout, httpx.ReadError):
38
+
pass
39
+
time.sleep(0.2 * (attempt + 1))
40
+
raise RuntimeError(f"gave up on {url} after 10 attempts")
41
+
TRANSITIONS = ["PENDING", "RUNNING"] # then 90% COMPLETED / 10% stay RUNNING
42
+
43
+
44
+
def seed(client: httpx.Client, base: str, n_runs: int, rng: random.Random) -> dict:
45
+
flow_ids = []
46
+
for i in range(FLOWS):
47
+
r = post_retry(client, f"{base}/flows/", {"name": f"bench-flow-{i}"})
48
+
flow_ids.append(r.json()["id"])
49
+
50
+
run_ids, completed = [], 0
51
+
for i in range(n_runs):
52
+
flow_id = flow_ids[rng.randrange(FLOWS)]
53
+
r = post_retry(client, f"{base}/flow_runs/", {
54
+
"flow_id": flow_id,
55
+
"name": f"bench-run-{i}",
56
+
"state": {"type": "PENDING"},
57
+
"tags": ["bench"],
58
+
})
59
+
run_id = r.json()["id"]
60
+
run_ids.append(run_id)
61
+
62
+
post_retry(client, f"{base}/flow_runs/{run_id}/set_state",
63
+
{"state": {"type": "RUNNING"}})
64
+
if rng.random() < 0.9:
65
+
post_retry(client, f"{base}/flow_runs/{run_id}/set_state",
66
+
{"state": {"type": "COMPLETED"}})
67
+
completed += 1
68
+
69
+
return {"flow_ids": flow_ids, "run_ids": run_ids, "completed": completed}
70
+
71
+
72
+
def patterns(base: str, seeded: dict, rng: random.Random):
73
+
run_ids = seeded["run_ids"]
74
+
return {
75
+
# cloud prototype: run-point-lookups.md
76
+
"point_lookup": lambda c: c.get(f"{base}/flow_runs/{rng.choice(run_ids)}"),
77
+
# cloud prototype: active_by_state (their p50: 14ms at whale scale)
78
+
"active_by_state": lambda c: c.post(
79
+
f"{base}/flow_runs/filter",
80
+
json={"flow_runs": {"state": {"type": {"any_": ["RUNNING"]}}}, "limit": 200},
81
+
),
82
+
# cloud prototype: terminal_completed_*
83
+
"completed_50": lambda c: c.post(
84
+
f"{base}/flow_runs/filter",
85
+
json={"flow_runs": {"state": {"type": {"any_": ["COMPLETED"]}}}, "limit": 50},
86
+
),
87
+
# cloud prototype: paginated_24h (OFFSET 100)
88
+
"paginated": lambda c: c.post(
89
+
f"{base}/flow_runs/filter",
90
+
json={
91
+
"flow_runs": {"state": {"type": {"any_": ["COMPLETED"]}}},
92
+
"limit": 50,
93
+
"offset": 100,
94
+
},
95
+
),
96
+
# cloud prototype: count via ZSET cardinality
97
+
"count_completed": lambda c: c.post(
98
+
f"{base}/flow_runs/count",
99
+
json={"flow_runs": {"state": {"type": {"any_": ["COMPLETED"]}}}},
100
+
),
101
+
# cloud prototype: deployment_list (closest entity-list shape)
102
+
"flow_list": lambda c: c.post(f"{base}/flows/", json={"name": "bench-flow-0"})
103
+
and c.post(f"{base}/flows/filter", json={"limit": 50}),
104
+
}
105
+
106
+
107
+
def measure(client, fn, iterations: int) -> dict:
108
+
samples = []
109
+
rows = None
110
+
for _ in range(iterations):
111
+
t0 = time.perf_counter()
112
+
r = fn(client)
113
+
elapsed = (time.perf_counter() - t0) * 1000
114
+
r.raise_for_status()
115
+
samples.append(elapsed)
116
+
if rows is None:
117
+
body = r.json()
118
+
rows = len(body) if isinstance(body, list) else 1
119
+
samples.sort()
120
+
return {
121
+
"p50": statistics.median(samples),
122
+
"p95": samples[int(len(samples) * 0.95) - 1],
123
+
"rows": rows,
124
+
}
125
+
126
+
127
+
def main():
128
+
ap = argparse.ArgumentParser()
129
+
ap.add_argument("--target", action="append", required=True, help="name=base_url")
130
+
ap.add_argument("--runs", type=int, default=300)
131
+
ap.add_argument("--iterations", type=int, default=30)
132
+
args = ap.parse_args()
133
+
134
+
results = {}
135
+
for spec in args.target:
136
+
name, base = spec.split("=", 1)
137
+
client = httpx.Client(timeout=60)
138
+
rng = random.Random(1337)
139
+
print(f"seeding {name}: {FLOWS} flows, {args.runs} runs ...", flush=True)
140
+
t0 = time.perf_counter()
141
+
seeded = seed(client, base, args.runs, rng)
142
+
seed_s = time.perf_counter() - t0
143
+
writes = FLOWS + args.runs * 2 + seeded["completed"]
144
+
print(f" seeded in {seed_s:.1f}s ({writes / seed_s:.0f} writes/s, "
145
+
f"{seeded['completed']} completed)", flush=True)
146
+
147
+
results[name] = {"seed_writes_per_s": writes / seed_s}
148
+
for pat, fn in patterns(base, seeded, random.Random(7)).items():
149
+
results[name][pat] = measure(client, fn, args.iterations)
150
+
m = results[name][pat]
151
+
print(f" {pat:20s} p50={m['p50']:7.2f}ms p95={m['p95']:7.2f}ms rows={m['rows']}")
152
+
153
+
# markdown table
154
+
names = list(results)
155
+
pats = [p for p in results[names[0]] if p != "seed_writes_per_s"]
156
+
print("\n| pattern | " + " | ".join(f"{n} p50/p95 (ms)" for n in names) + " |")
157
+
print("|---|" + "---|" * len(names))
158
+
print("| seed write rate | " + " | ".join(
159
+
f"{results[n]['seed_writes_per_s']:.0f}/s" for n in names) + " |")
160
+
for p in pats:
161
+
cells = [f"{results[n][p]['p50']:.1f} / {results[n][p]['p95']:.1f}" for n in names]
162
+
print(f"| {p} | " + " | ".join(cells) + " |")
163
+
164
+
print(json.dumps(results, indent=2, default=float), file=open("/tmp/bench-reads.json", "w"))
165
+
166
+
167
+
if __name__ == "__main__":
168
+
main()
+295
-8
src/api.zig
+295
-8
src/api.zig
···
68
68
if (std.mem.eql(u8, path, "/flows/")) return self.createFlow(req, res);
69
69
if (std.mem.eql(u8, path, "/flow_runs/")) return self.createFlowRun(req, res);
70
70
if (std.mem.eql(u8, path, "/task_runs/")) return self.createTaskRun(req, res);
71
+
if (std.mem.eql(u8, path, "/flows/filter")) return self.filterFlows(req, res);
72
+
if (std.mem.eql(u8, path, "/flow_runs/filter")) return self.filterRuns(req, res, .flow_run, false);
73
+
if (std.mem.eql(u8, path, "/flow_runs/count")) return self.filterRuns(req, res, .flow_run, true);
74
+
if (std.mem.eql(u8, path, "/task_runs/filter")) return self.filterRuns(req, res, .task_run, false);
75
+
if (std.mem.eql(u8, path, "/task_runs/count")) return self.filterRuns(req, res, .task_run, true);
71
76
if (setStateId(path, "/flow_runs/")) |id| return self.setState(req, res, id, .flow_run);
72
77
if (setStateId(path, "/task_runs/")) |id| return self.setState(req, res, id, .task_run);
73
78
if (std.mem.eql(u8, path, "/logs/")) return self.createLogs(req, res);
···
232
237
233
238
try self.cache.set(try cacheKey(arena, "flow", id), flow_json);
234
239
try self.cache.set(name_key, id);
240
+
try self.cache.sadd("s:flow:all", id);
235
241
236
242
res.setStatus(.created);
237
243
res.body = try arena.dupe(u8, flow_json);
···
289
295
try self.writeRecord(arena, flow_run_collection, row.id, record_json);
290
296
291
297
try self.storeRow(arena, "flow_run", row);
298
+
try self.indexRun(arena, "flow_run", row);
292
299
293
300
res.setStatus(.created);
294
301
try self.writeRunResponse(res, .flow_run, row);
···
337
344
try self.writeRecord(arena, task_run_collection, row.id, record_json);
338
345
339
346
try self.storeRow(arena, "task_run", row);
347
+
try self.indexRun(arena, "task_run", row);
340
348
341
349
res.setStatus(.created);
342
350
try self.writeRunResponse(res, .task_run, row);
343
351
}
344
352
353
+
/// secondary indexes: membership sets per state / per parent, mirroring
354
+
/// the cloud prototype's redis run-state indices. derived, rebuildable.
355
+
fn indexRun(self: *Api, arena: std.mem.Allocator, kind_str: []const u8, row: RunRow) !void {
356
+
try self.cache.sadd(try std.fmt.allocPrint(arena, "s:{s}:all", .{kind_str}), row.id);
357
+
if (row.state_id.len > 0) {
358
+
try self.cache.sadd(try std.fmt.allocPrint(arena, "s:{s}:state:{s}", .{ kind_str, row.state_type }), row.id);
359
+
}
360
+
if (row.flow_id.len > 0) {
361
+
try self.cache.sadd(try std.fmt.allocPrint(arena, "s:{s}:flow:{s}", .{ kind_str, row.flow_id }), row.id);
362
+
}
363
+
if (row.flow_run_id) |frid| {
364
+
try self.cache.sadd(try std.fmt.allocPrint(arena, "s:{s}:flowrun:{s}", .{ kind_str, frid }), row.id);
365
+
}
366
+
}
367
+
368
+
fn moveStateIndex(self: *Api, arena: std.mem.Allocator, kind_str: []const u8, id: []const u8, prior: ?[]const u8, new_type: []const u8) !void {
369
+
if (prior) |p| {
370
+
try self.cache.srem(try std.fmt.allocPrint(arena, "s:{s}:state:{s}", .{ kind_str, p }), id);
371
+
}
372
+
try self.cache.sadd(try std.fmt.allocPrint(arena, "s:{s}:state:{s}", .{ kind_str, new_type }), id);
373
+
}
374
+
345
375
/// runs are created with an optional initial state (the client sends
346
376
/// Pending). recorded like any other transition: an event.
347
377
fn applyInitialState(self: *Api, arena: std.mem.Allocator, row: *RunRow, parsed: std.json.Value, now: []const u8, kind: RunKind) !void {
···
422
452
try self.writeTransitionEvent(arena, kind, row, prior_type);
423
453
424
454
try self.storeRow(arena, kind_str, row);
455
+
try self.moveStateIndex(arena, kind_str, row.id, prior_type, row.state_type);
425
456
426
457
// OrchestrationResult: state + ACCEPT
427
458
var out: std.Io.Writer.Allocating = .init(arena);
···
506
537
_ = self;
507
538
var out: std.Io.Writer.Allocating = .init(res.arena);
508
539
var jw: std.json.Stringify = .{ .writer = &out.writer };
540
+
try writeRunObject(&jw, kind, row);
541
+
res.body = out.written();
542
+
}
509
543
544
+
fn writeRunObject(jw: *std.json.Stringify, kind: RunKind, row: RunRow) !void {
510
545
try jw.beginObject();
511
546
try jw.objectField("id");
512
547
try jw.write(row.id);
···
522
557
try jw.objectField("flow_id");
523
558
try jw.write(row.flow_id);
524
559
try jw.objectField("parameters");
525
-
try writeRaw(&jw, row.raw_parameters);
560
+
try writeRaw(jw, row.raw_parameters);
526
561
try jw.objectField("job_variables");
527
-
try writeRaw(&jw, row.raw_job_variables);
562
+
try writeRaw(jw, row.raw_job_variables);
528
563
try jw.objectField("deployment_id");
529
564
try jw.write(null);
530
565
try jw.objectField("deployment_version");
···
554
589
try jw.objectField("task_version");
555
590
try jw.write(row.task_version);
556
591
try jw.objectField("task_inputs");
557
-
try writeRaw(&jw, row.raw_task_inputs);
592
+
try writeRaw(jw, row.raw_task_inputs);
558
593
},
559
594
}
560
595
561
596
try jw.objectField("tags");
562
-
try writeRaw(&jw, row.raw_tags);
597
+
try writeRaw(jw, row.raw_tags);
563
598
try jw.objectField("empirical_policy");
564
-
try writeRaw(&jw, row.raw_empirical_policy);
599
+
try writeRaw(jw, row.raw_empirical_policy);
565
600
try jw.objectField("run_count");
566
601
try jw.write(row.run_count);
567
602
try jw.objectField("expected_start_time");
···
578
613
try jw.write(if (row.state_id.len > 0) row.state_name else null);
579
614
try jw.objectField("state");
580
615
if (row.state_id.len > 0) {
581
-
try writeStateObject(&jw, row, kind == .task_run);
616
+
try writeStateObject(jw, row, kind == .task_run);
582
617
} else {
583
618
try jw.write(null);
584
619
}
585
620
try jw.endObject();
586
-
587
-
res.body = out.written();
588
621
}
589
622
590
623
fn writeStateObject(jw: *std.json.Stringify, row: RunRow, include_data: bool) !void {
···
606
639
try jw.write(null);
607
640
}
608
641
try jw.endObject();
642
+
}
643
+
644
+
// ------------------------------------------------------------------
645
+
// filtered reads — served entirely from the appview. candidate ids
646
+
// come from membership sets; rows sort by ISO timestamp (string order
647
+
// == time order); limit/offset applied after sort.
648
+
// ------------------------------------------------------------------
649
+
650
+
fn filterRuns(self: *Api, req: *httpz.Request, res: *httpz.Response, kind: RunKind, count_only: bool) !void {
651
+
const arena = req.arena;
652
+
const kind_str = switch (kind) {
653
+
.flow_run => "flow_run",
654
+
.task_run => "task_run",
655
+
};
656
+
657
+
const parsed: std.json.Value = blk: {
658
+
const body = req.body() orelse break :blk .null;
659
+
if (body.len == 0) break :blk .null;
660
+
break :blk std.json.parseFromSliceLeaky(std.json.Value, arena, body, .{}) catch .null;
661
+
};
662
+
663
+
const runs_filter: std.json.Value = blk: {
664
+
if (parsed != .object) break :blk .null;
665
+
const key = switch (kind) {
666
+
.flow_run => "flow_runs",
667
+
.task_run => "task_runs",
668
+
};
669
+
break :blk parsed.object.get(key) orelse .null;
670
+
};
671
+
672
+
// candidates: state filter -> union of state sets, else the all-set
673
+
var candidates: std.ArrayList([]const u8) = .empty;
674
+
const state_types = filterAny(runs_filter, "state", "type");
675
+
if (state_types) |types| {
676
+
for (types) |t| {
677
+
if (t != .string) continue;
678
+
const members = try self.cache.smembers(arena, try std.fmt.allocPrint(arena, "s:{s}:state:{s}", .{ kind_str, t.string }));
679
+
try candidates.appendSlice(arena, members);
680
+
}
681
+
} else {
682
+
const members = try self.cache.smembers(arena, try std.fmt.allocPrint(arena, "s:{s}:all", .{kind_str}));
683
+
try candidates.appendSlice(arena, members);
684
+
}
685
+
686
+
// intersect with flow-run scope for task runs (the engine's
687
+
// "task runs of this flow run" read)
688
+
if (kind == .task_run) {
689
+
if (filterAny(runs_filter, "flow_run_id", null)) |frids| {
690
+
if (frids.len > 0 and frids[0] == .string) {
691
+
const scoped = try self.cache.smembers(arena, try std.fmt.allocPrint(arena, "s:task_run:flowrun:{s}", .{frids[0].string}));
692
+
candidates.items = try intersect(arena, candidates.items, scoped);
693
+
}
694
+
}
695
+
}
696
+
697
+
if (count_only) {
698
+
res.body = try std.fmt.allocPrint(arena, "{d}", .{candidates.items.len});
699
+
return;
700
+
}
701
+
702
+
// load rows, sort newest-first, slice
703
+
var rows: std.ArrayList(RunRow) = .empty;
704
+
for (candidates.items) |id| {
705
+
if (try self.loadRow(arena, kind_str, id)) |row| try rows.append(arena, row);
706
+
}
707
+
std.mem.sort(RunRow, rows.items, {}, struct {
708
+
fn newerFirst(_: void, a: RunRow, b: RunRow) bool {
709
+
return std.mem.order(u8, a.created, b.created) == .gt;
710
+
}
711
+
}.newerFirst);
712
+
713
+
const offset: usize = if (parsed == .object)
714
+
if (parsed.object.get("offset")) |o| (if (o == .integer and o.integer > 0) @intCast(o.integer) else 0) else 0
715
+
else
716
+
0;
717
+
const limit: usize = if (parsed == .object)
718
+
if (parsed.object.get("limit")) |l| (if (l == .integer and l.integer > 0) @intCast(l.integer) else 200) else 200
719
+
else
720
+
200;
721
+
722
+
const start = @min(offset, rows.items.len);
723
+
const end = @min(start + limit, rows.items.len);
724
+
725
+
var out: std.Io.Writer.Allocating = .init(arena);
726
+
var jw: std.json.Stringify = .{ .writer = &out.writer };
727
+
try jw.beginArray();
728
+
for (rows.items[start..end]) |row| try writeRunObject(&jw, kind, row);
729
+
try jw.endArray();
730
+
res.body = out.written();
731
+
}
732
+
733
+
fn filterFlows(self: *Api, req: *httpz.Request, res: *httpz.Response) !void {
734
+
const arena = req.arena;
735
+
const ids = try self.cache.smembers(arena, "s:flow:all");
736
+
737
+
const Named = struct { name: []const u8, raw: []const u8 };
738
+
var flows: std.ArrayList(Named) = .empty;
739
+
for (ids) |id| {
740
+
const raw = self.cache.get(arena, try cacheKey(arena, "flow", id)) orelse continue;
741
+
const v = std.json.parseFromSliceLeaky(std.json.Value, arena, raw, .{}) catch continue;
742
+
const name = getString(v, "name") orelse "";
743
+
try flows.append(arena, .{ .name = name, .raw = raw });
744
+
}
745
+
std.mem.sort(Named, flows.items, {}, struct {
746
+
fn byName(_: void, a: Named, b: Named) bool {
747
+
return std.mem.order(u8, a.name, b.name) == .lt;
748
+
}
749
+
}.byName);
750
+
751
+
var out: std.Io.Writer.Allocating = .init(arena);
752
+
var jw: std.json.Stringify = .{ .writer = &out.writer };
753
+
try jw.beginArray();
754
+
for (flows.items) |f| {
755
+
try jw.beginWriteRaw();
756
+
try jw.writer.writeAll(f.raw);
757
+
jw.endWriteRaw();
758
+
}
759
+
try jw.endArray();
760
+
res.body = out.written();
761
+
}
762
+
763
+
/// navigate filter json like flow_runs.state.type.any_ -> []Value
764
+
fn filterAny(filter: std.json.Value, key: []const u8, sub: ?[]const u8) ?[]std.json.Value {
765
+
if (filter != .object) return null;
766
+
var node = filter.object.get(key) orelse return null;
767
+
if (sub) |s| {
768
+
if (node != .object) return null;
769
+
node = node.object.get(s) orelse return null;
770
+
}
771
+
if (node != .object) return null;
772
+
const any = node.object.get("any_") orelse return null;
773
+
if (any != .array) return null;
774
+
return any.array.items;
775
+
}
776
+
777
+
fn intersect(arena: std.mem.Allocator, a: []const []const u8, b: [][]u8) ![][]const u8 {
778
+
var keep: std.StringHashMapUnmanaged(void) = .empty;
779
+
for (b) |m| try keep.put(arena, m, {});
780
+
var out: std.ArrayList([]const u8) = .empty;
781
+
for (a) |m| if (keep.contains(m)) try out.append(arena, m);
782
+
return out.items;
783
+
}
784
+
785
+
// ------------------------------------------------------------------
786
+
// hydration: rebuild the entire appview by replaying the repo.
787
+
// entities first, then the event log in TID order. this is the
788
+
// "delete the cache, replay the truth" property, exercised on boot.
789
+
// ------------------------------------------------------------------
790
+
791
+
pub fn hydrate(self: *Api, arena: std.mem.Allocator) !usize {
792
+
var applied: usize = 0;
793
+
794
+
for (try self.repo.listRecords(arena, flow_collection)) |rec| {
795
+
const v = zat.cbor.decodeAll(arena, rec.cbor_data) catch continue;
796
+
const id = rec.key[flow_collection.len + 1 ..];
797
+
const name = v.getString("name") orelse continue;
798
+
const created = v.getString("created") orelse "";
799
+
const flow_json = try std.fmt.allocPrint(arena,
800
+
\\{{"id":"{s}","created":"{s}","updated":"{s}","name":"{s}","tags":{s}}}
801
+
, .{ id, created, created, name, try cborFieldJson(arena, v, "tags", "[]") });
802
+
try self.cache.set(try cacheKey(arena, "flow", id), flow_json);
803
+
try self.cache.set(try std.fmt.allocPrint(arena, "flow:name:{s}", .{name}), id);
804
+
try self.cache.sadd("s:flow:all", id);
805
+
applied += 1;
806
+
}
807
+
808
+
for (try self.repo.listRecords(arena, flow_run_collection)) |rec| {
809
+
const v = zat.cbor.decodeAll(arena, rec.cbor_data) catch continue;
810
+
const id = rec.key[flow_run_collection.len + 1 ..];
811
+
const row = RunRow{
812
+
.id = id,
813
+
.created = v.getString("created") orelse "",
814
+
.updated = v.getString("created") orelse "",
815
+
.name = v.getString("name") orelse "",
816
+
.flow_id = v.getString("flowId") orelse "",
817
+
.raw_parameters = try cborFieldJson(arena, v, "parameters", "{}"),
818
+
.raw_tags = try cborFieldJson(arena, v, "tags", "[]"),
819
+
};
820
+
try self.storeRow(arena, "flow_run", row);
821
+
try self.indexRun(arena, "flow_run", row);
822
+
applied += 1;
823
+
}
824
+
825
+
for (try self.repo.listRecords(arena, task_run_collection)) |rec| {
826
+
const v = zat.cbor.decodeAll(arena, rec.cbor_data) catch continue;
827
+
const id = rec.key[task_run_collection.len + 1 ..];
828
+
const row = RunRow{
829
+
.id = id,
830
+
.created = v.getString("created") orelse "",
831
+
.updated = v.getString("created") orelse "",
832
+
.name = v.getString("name") orelse "",
833
+
.flow_run_id = v.getString("flowRunId"),
834
+
.task_key = v.getString("taskKey") orelse "",
835
+
.dynamic_key = v.getString("dynamicKey") orelse "",
836
+
};
837
+
try self.storeRow(arena, "task_run", row);
838
+
try self.indexRun(arena, "task_run", row);
839
+
applied += 1;
840
+
}
841
+
842
+
// replay the event log: MST walk order == TID order == time order
843
+
for (try self.repo.listRecords(arena, event_collection)) |rec| {
844
+
const v = zat.cbor.decodeAll(arena, rec.cbor_data) catch continue;
845
+
const event_name = v.getString("event") orelse continue;
846
+
const resource = v.get("resource") orelse continue;
847
+
const resource_id = resource.getString("prefect.resource.id") orelse continue;
848
+
849
+
const kind_str: []const u8, const kind: RunKind, const prefix: []const u8 =
850
+
if (std.mem.indexOf(u8, event_name, "flow-run") != null)
851
+
.{ "flow_run", .flow_run, "prefect.flow-run." }
852
+
else if (std.mem.indexOf(u8, event_name, "task-run") != null)
853
+
.{ "task_run", .task_run, "prefect.task-run." }
854
+
else
855
+
continue;
856
+
_ = kind;
857
+
if (!std.mem.startsWith(u8, resource_id, prefix)) continue;
858
+
const run_id = resource_id[prefix.len..];
859
+
860
+
const payload = v.get("payload") orelse continue;
861
+
const state = payload.get("validated_state") orelse continue;
862
+
const state_type = state.getString("type") orelse continue;
863
+
const ts = state.getString("timestamp") orelse "";
864
+
865
+
var row = (try self.loadRow(arena, kind_str, run_id)) orelse continue;
866
+
const prior: ?[]const u8 = if (row.state_id.len > 0) try arena.dupe(u8, row.state_type) else null;
867
+
868
+
row.state_id = v.getString("eventId") orelse "";
869
+
row.state_type = try arena.dupe(u8, state_type);
870
+
row.state_name = try arena.dupe(u8, state.getString("name") orelse defaultStateName(state_type));
871
+
row.state_timestamp = ts;
872
+
row.state_message = state.getString("message");
873
+
row.raw_state_details = try cborFieldJson(arena, state, "stateDetails", "{}");
874
+
row.updated = ts;
875
+
if (std.mem.eql(u8, state_type, "RUNNING")) {
876
+
row.run_count += 1;
877
+
if (row.start_time == null) row.start_time = ts;
878
+
}
879
+
if (isTerminal(state_type)) row.end_time = ts;
880
+
881
+
try self.storeRow(arena, kind_str, row);
882
+
try self.moveStateIndex(arena, kind_str, row.id, prior, row.state_type);
883
+
applied += 1;
884
+
}
885
+
886
+
return applied;
887
+
}
888
+
889
+
/// serialize a cbor map field back to JSON text (or default)
890
+
fn cborFieldJson(arena: std.mem.Allocator, v: zat.cbor.Value, key: []const u8, default: []const u8) ![]const u8 {
891
+
const sub = v.get(key) orelse return default;
892
+
var out: std.Io.Writer.Allocating = .init(arena);
893
+
var jw: std.json.Stringify = .{ .writer = &out.writer };
894
+
try json_cbor.cborToJson(&jw, sub);
895
+
return out.written();
609
896
}
610
897
611
898
// ------------------------------------------------------------------
+28
src/cache.zig
+28
src/cache.zig
···
36
36
.burner => |*b| b.store.get(out_alloc, key) catch null,
37
37
};
38
38
}
39
+
40
+
// set ops back the appview's secondary indexes (runs by state, by flow)
41
+
// — the same shape as the cloud prototype's redis run-state indices.
42
+
43
+
pub fn sadd(self: *Cache, key: []const u8, member: []const u8) !void {
44
+
switch (self.*) {
45
+
.burner => |*b| _ = try b.store.sadd(key, &.{member}),
46
+
}
47
+
}
48
+
49
+
pub fn srem(self: *Cache, key: []const u8, member: []const u8) !void {
50
+
switch (self.*) {
51
+
.burner => |*b| _ = b.store.srem(key, &.{member}) catch 0,
52
+
}
53
+
}
54
+
55
+
/// returned members owned by out_alloc
56
+
pub fn smembers(self: *Cache, out_alloc: std.mem.Allocator, key: []const u8) ![][]u8 {
57
+
return switch (self.*) {
58
+
.burner => |*b| b.store.smembers(out_alloc, key) catch &.{},
59
+
};
60
+
}
61
+
62
+
pub fn scard(self: *Cache, key: []const u8) i64 {
63
+
return switch (self.*) {
64
+
.burner => |*b| b.store.scard(key) catch 0,
65
+
};
66
+
}
39
67
};
40
68
41
69
const Burner = struct {
+4
-1
src/main.zig
+4
-1
src/main.zig
···
46
46
log.info("workspace did: {s}", .{repo.did});
47
47
log.info("repo: {d} records, data dir: {s}", .{ repo.record_count, data_dir });
48
48
if (repo.record_count > 0) {
49
-
log.warn("appview hydration from existing CAR not implemented yet — reads of prior entities will 404 (records remain intact in the repo)", .{});
49
+
var hydrate_arena = std.heap.ArenaAllocator.init(allocator);
50
+
defer hydrate_arena.deinit();
51
+
const applied = try api.hydrate(hydrate_arena.allocator());
52
+
log.info("appview hydrated: {d} records replayed from the repo", .{applied});
50
53
}
51
54
52
55
const ip = try std.Io.net.IpAddress.parse("127.0.0.1", port);