zzstoatzz.io
This repository has no description
scale: real-redis appview, commit cadence, and the 8000-run gauntlet
production shape lands: the Cache union gains a real-redis-over-TCP
backend (PREFECT_PDS_CACHE=redis, container via orbstack), selected at
boot; burner stays the zero-infra default. commit cadence becomes a
knob (PREFECT_PDS_COMMIT_EVERY) — the same lever as the cloud
prototype's delta-flush cadence.
scripts/bench-scale drives 16 concurrent client lifecycles in
cumulative tiers and found three real bugs before producing numbers:
quadratic CAR retention here (fixed: per-commit scratch arena +
child-allocator-owned latest_car), quadratic MST re-serialization in
zat (fixed upstream on zat#mst-block-cache), and realloc-dangled
response slices in the redis client under large SMEMBERS (fixed
upstream on redis#retired-read-buffers).
results (docs/scale-results.md): 8000 runs / ~30k records, no crashes.
point reads flat at ~1ms across 32x scale; per-write commits decay
O(records) as predicted; commit_every=20 flattens throughput at
840 w/s with p95 *improving* at depth. remaining walls are slopes with
named fixes, not cliffs.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
author
zzstoatzz
co-author
Claude Fable 5
date
(Jun 12, 2026, 3:32 AM -0500)
commit
82925b4b
82925b4b24340f170a5f9d063484001ef0468b11
parent
41fadeda
41fadeda27c2f222ec487be9a065455fa3286aa9
+380
-26
+2
build.zig
+2
build.zig
···
7
7
const zat_dep = b.dependency("zat", .{ .target = target, .optimize = optimize });
8
8
const httpz_dep = b.dependency("httpz", .{ .target = target, .optimize = optimize });
9
9
const burner_dep = b.dependency("burner_redis", .{ .target = target, .optimize = optimize });
10
+
const redis_dep = b.dependency("redis", .{ .target = target, .optimize = optimize });
10
11
11
12
const imports: []const std.Build.Module.Import = &.{
12
13
.{ .name = "zat", .module = zat_dep.module("zat") },
13
14
.{ .name = "httpz", .module = httpz_dep.module("httpz") },
14
15
.{ .name = "burner_redis", .module = burner_dep.module("burner_redis") },
16
+
.{ .name = "redis", .module = redis_dep.module("redis") },
15
17
};
16
18
17
19
const exe = b.addExecutable(.{
+8
-4
build.zig.zon
+8
-4
build.zig.zon
···
5
5
.minimum_zig_version = "0.16.0",
6
6
.dependencies = .{
7
7
.zat = .{
8
-
.url = "https://tangled.org/zat.dev/zat/archive/v0.3.6.tar.gz",
9
-
.hash = "zat-0.3.6-5PuC7l5xCQDpuYx8mdk22znf8GntN7gv9372YtIDVtzy",
8
+
.url = "git+https://tangled.org/zat.dev/zat?ref=mst-block-cache#fa178ee5136859a89c4643d1fbe74c9ff41a430e",
9
+
.hash = "zat-0.3.6-5PuC7p11CQBmYWRtyxRUgXOcetSSHLDRnD5GjZStgeYn",
10
10
},
11
11
.httpz = .{
12
-
.url = "git+https://tangled.org/zzstoatzz.io/http.zig?ref=zat-websocket-pin#2e050036487b2e6638e9ac3d315bf1de85dd849f",
13
-
.hash = "httpz-0.0.0-PNVzrEvRCABghgb-wCOwrFNxgiDKtHgXW4O_J2DWvjTl",
12
+
.url = "git+https://tangled.org/zzstoatzz.io/http.zig?ref=zat-websocket-pin#7f415c75f7d7ba1adf608b0b074d553221ab6748",
13
+
.hash = "httpz-0.0.0-PNVzrLzSCACt2P5jixSdJ5Gro35zic4hb40S1B3CsoNm",
14
14
},
15
15
.burner_redis = .{
16
16
.url = "https://tangled.org/zzstoatzz.io/burner-redis/archive/fdb749a.tar.gz",
17
17
.hash = "burner_redis_zig-0.0.4-JKoFUPaOBAA_0SSPM9pTk6WGVv1rnRLpSJG938zpOsq_",
18
+
},
19
+
.redis = .{
20
+
.url = "https://tangled.org/zzstoatzz.io/redis/archive/507f69f682ce3d1718652a6d48706d9eec485a98.tar.gz",
21
+
.hash = "redis-0.1.0-NGPovupWAgDCCiDd5w6S9N7Jzyb06XVa91hHD-CSGpAN",
18
22
},
19
23
},
20
24
.paths = .{
+76
docs/scale-results.md
+76
docs/scale-results.md
···
1
+
# scale results — 2026-06-12
2
+
3
+
`scripts/bench-scale`: 16 concurrent client lifecycles (create → Running →
4
+
Completed) through the REST API, cumulative tiers, reads measured at each
5
+
tier boundary. **production-shaped stack**: ReleaseFast server, appview in a
6
+
real redis 7 container (orbstack), repo CAR on disk. each run ≈ 3.7 records
7
+
(flowRun + 2-3 events), so tier 8000 ≈ 29,600 records.
8
+
9
+
## per-write commits (commit_every=1, strictest durability)
10
+
11
+
| total runs | write p50/p95 (ms) | writes/s | point | active | completed_50 | count | CAR MB | RSS MB |
12
+
|---|---|---|---|---|---|---|---|---|
13
+
| 250 | 12.0 / 53.8 | 855 | 1.03 | 3.3 | 16.1 | 1.02 | 0.4 | 16 |
14
+
| 500 | 14.3 / 62.7 | 700 | 0.89 | 6.3 | 29.7 | 1.02 | 0.9 | 24 |
15
+
| 1000 | 17.3 / 89.1 | 560 | 0.97 | 9.6 | 71.4 | 1.01 | 1.8 | 38 |
16
+
| 2000 | 20.8 / 124.7 | 404 | 1.07 | 19.2 | 122.0 | 1.11 | 3.6 | 76 |
17
+
| 4000 | 37.5 / 231.7 | 227 | 1.10 | 28.7 | 242.4 | 1.24 | 7.2 | 152 |
18
+
| 8000 | 27.7 / 232.9 | 118 | 1.03 | 55.1 | 476.5 | 0.93 | 14.4 | 308 |
19
+
20
+
## batched commits (commit_every=20 — the cadence lever)
21
+
22
+
| total runs | write p50/p95 (ms) | writes/s | point | active | completed_50 | count | RSS MB |
23
+
|---|---|---|---|---|---|---|---|
24
+
| 1000 | 14.5 / 64.2 | 702 | 0.97 | 7.3 | 58.9 | 0.83 | 22 |
25
+
| 4000 | 13.7 / 61.9 | 729 | 0.98 | 32.4 | 242.6 | 0.97 | 57 |
26
+
| 8000 | **13.2 / 51.9** | **840** | 1.00 | 54.1 | 484.9 | 0.89 | 104 |
27
+
28
+
## what the curves say
29
+
30
+
- **point lookups and counts are flat at ~1ms across a 32× scale increase.**
31
+
the O(1) appview promise holds under concurrent load against real redis.
32
+
- **per-write commits decay as O(records)** — 855 → 118 w/s — because every
33
+
commit rewrites the full CAR (0.4 → 14.4 MB). this was the predicted
34
+
known-naive wall, now measured.
35
+
- **commit cadence flattens it completely**: at every-20-writes, throughput
36
+
is scale-independent (840 w/s at tier 8000, p95 *improving*). this is the
37
+
cloud prototype's delta-flush-cadence finding reproduced in our
38
+
architecture — their `scan_redis_deltas` lever, our `commit_every`. the
39
+
durability window (≤19 uncommitted writes) is the price; block-level CAR
40
+
append is the design that removes the trade entirely.
41
+
- **filtered lists degrade O(candidates)** — completed_50 16 → 477ms —
42
+
because the filter engine loads all candidates then sorts and slices.
43
+
predicted in access-patterns.md; the fix (created-time zset so the slice
44
+
happens before row loads) is mechanical.
45
+
- **memory is linear now**: ~10KB/record RSS. it was quadratic twice over
46
+
before this run (see below).
47
+
48
+
## bugs this harness found (both upstream, both fixed)
49
+
50
+
1. **zat: quadratic MST allocations.** `collectBlocks` re-serialized every
51
+
clean node on every commit — multi-GB RSS by a few thousand records.
52
+
fixed on `zat#mst-block-cache`: clean nodes cache their encoding (the
53
+
dirty flag already gates validity); plus `collectBlocksInto(gpa, out)`
54
+
so per-commit lists grow in scratch memory. 455/455 zat tests pass,
55
+
including interop vectors.
56
+
2. **redis (zig client): realloc dangled parsed slices.** a response larger
57
+
than the read buffer grew it with `realloc`, moving the allocation and
58
+
dangling every bulk slice already parsed — segfault on a ~1800-member
59
+
SMEMBERS. fixed on `redis#retired-read-buffers`: growth copies to a
60
+
fresh buffer and retires the old one until the next command, which is
61
+
precisely the documented slice lifetime.
62
+
63
+
also fixed locally: prefect-pds retained every full CAR ever written in the
64
+
repo arena (one per commit — quadratic). commit-transient memory now lives
65
+
in a per-commit scratch arena and `latest_car` is freed on replacement.
66
+
67
+
## where it actually breaks now
68
+
69
+
nothing crashes through 8,000 runs / ~30k records. the remaining walls are
70
+
performance slopes, not cliffs, and each has a named fix:
71
+
72
+
| slope | cause | fix |
73
+
|---|---|---|
74
+
| write decay (cadence=1) | full CAR rewrite per commit | block-level CAR append |
75
+
| filter latency | load-all-candidates | time-ordered zset + early slice |
76
+
| RSS growth | record bytes held twice (repo blocks + appview rows) | appview rows already live in redis; trim the in-process copy |
+151
scripts/bench-scale
+151
scripts/bench-scale
···
1
+
#!/usr/bin/env -S uv run --script --quiet
2
+
# /// script
3
+
# requires-python = ">=3.12"
4
+
# dependencies = ["httpx"]
5
+
# ///
6
+
"""find where it breaks: concurrent client load at increasing scale.
7
+
8
+
drives C concurrent run lifecycles (create -> Running -> Completed) through
9
+
the REST API in cumulative tiers, measuring at each tier boundary:
10
+
- write latency p50/p95 over the tier (per request)
11
+
- write throughput over the tier
12
+
- read pattern p50s at current scale
13
+
- repo CAR size on disk + server RSS
14
+
15
+
the expected failure mode is the full-CAR-rewrite-per-commit: write cost
16
+
should grow with total record count. this harness exists to draw that curve.
17
+
18
+
usage:
19
+
./scripts/bench-scale [--base http://localhost:4200/api] \
20
+
[--tiers 250,500,1000,2000,4000] [--concurrency 16] [--data-dir data]
21
+
"""
22
+
import argparse
23
+
import asyncio
24
+
import statistics
25
+
import subprocess
26
+
import time
27
+
from pathlib import Path
28
+
29
+
import httpx
30
+
31
+
32
+
async def lifecycle(client, base, flow_id, i, samples):
33
+
async def post(url, body):
34
+
t0 = time.perf_counter()
35
+
r = await client.post(url, json=body)
36
+
samples.append((time.perf_counter() - t0) * 1000)
37
+
r.raise_for_status()
38
+
return r
39
+
40
+
r = await post(f"{base}/flow_runs/", {
41
+
"flow_id": flow_id,
42
+
"name": f"scale-run-{i}",
43
+
"state": {"type": "PENDING"},
44
+
})
45
+
run_id = r.json()["id"]
46
+
await post(f"{base}/flow_runs/{run_id}/set_state", {"state": {"type": "RUNNING"}})
47
+
if i % 10 != 0: # leave 10% running
48
+
await post(f"{base}/flow_runs/{run_id}/set_state", {"state": {"type": "COMPLETED"}})
49
+
return run_id
50
+
51
+
52
+
async def read_patterns(client, base, sample_run_id):
53
+
out = {}
54
+
for name, method, url, body in [
55
+
("point_lookup", "GET", f"{base}/flow_runs/{sample_run_id}", None),
56
+
("active_by_state", "POST", f"{base}/flow_runs/filter",
57
+
{"flow_runs": {"state": {"type": {"any_": ["RUNNING"]}}}, "limit": 200}),
58
+
("completed_50", "POST", f"{base}/flow_runs/filter",
59
+
{"flow_runs": {"state": {"type": {"any_": ["COMPLETED"]}}}, "limit": 50}),
60
+
("count_completed", "POST", f"{base}/flow_runs/count",
61
+
{"flow_runs": {"state": {"type": {"any_": ["COMPLETED"]}}}}),
62
+
]:
63
+
samples = []
64
+
for _ in range(15):
65
+
t0 = time.perf_counter()
66
+
if method == "GET":
67
+
r = await client.get(url)
68
+
else:
69
+
r = await client.post(url, json=body)
70
+
samples.append((time.perf_counter() - t0) * 1000)
71
+
r.raise_for_status()
72
+
out[name] = statistics.median(samples)
73
+
return out
74
+
75
+
76
+
def server_stats(data_dir: str) -> dict:
77
+
car = Path(data_dir) / "repo.car"
78
+
car_mb = car.stat().st_size / 1048576 if car.exists() else 0
79
+
rss_mb = 0.0
80
+
try:
81
+
pid = subprocess.run(["pgrep", "-f", "bin/prefect-pds"],
82
+
capture_output=True, text=True).stdout.split()[0]
83
+
rss_kb = subprocess.run(["ps", "-o", "rss=", "-p", pid],
84
+
capture_output=True, text=True).stdout.strip()
85
+
rss_mb = int(rss_kb) / 1024
86
+
except (IndexError, ValueError):
87
+
pass
88
+
return {"car_mb": car_mb, "rss_mb": rss_mb}
89
+
90
+
91
+
async def main():
92
+
ap = argparse.ArgumentParser()
93
+
ap.add_argument("--base", default="http://localhost:4200/api")
94
+
ap.add_argument("--tiers", default="250,500,1000,2000,4000")
95
+
ap.add_argument("--concurrency", type=int, default=16)
96
+
ap.add_argument("--data-dir", default="data")
97
+
args = ap.parse_args()
98
+
tiers = [int(t) for t in args.tiers.split(",")]
99
+
100
+
async with httpx.AsyncClient(timeout=120) as client:
101
+
r = await client.post(f"{args.base}/flows/", json={"name": "scale-flow"})
102
+
r.raise_for_status()
103
+
flow_id = r.json()["id"]
104
+
105
+
total, rows = 0, []
106
+
sem = asyncio.Semaphore(args.concurrency)
107
+
last_run_id = None
108
+
109
+
for tier in tiers:
110
+
n_new = tier - total
111
+
samples: list[float] = []
112
+
113
+
async def guarded(i, samples=samples):
114
+
async with sem:
115
+
return await lifecycle(client, args.base, flow_id, i, samples)
116
+
117
+
t0 = time.perf_counter()
118
+
run_ids = await asyncio.gather(*(guarded(total + i) for i in range(n_new)))
119
+
elapsed = time.perf_counter() - t0
120
+
last_run_id = run_ids[-1]
121
+
total = tier
122
+
123
+
samples.sort()
124
+
reads = await read_patterns(client, args.base, last_run_id)
125
+
stats = server_stats(args.data_dir)
126
+
row = {
127
+
"tier": tier,
128
+
"write_p50": statistics.median(samples),
129
+
"write_p95": samples[int(len(samples) * 0.95) - 1],
130
+
"writes_per_s": len(samples) / elapsed,
131
+
**reads,
132
+
**stats,
133
+
}
134
+
rows.append(row)
135
+
print(f"tier {tier:6d}: write p50={row['write_p50']:6.1f}ms "
136
+
f"p95={row['write_p95']:7.1f}ms {row['writes_per_s']:6.0f} w/s | "
137
+
f"point={row['point_lookup']:5.2f}ms active={row['active_by_state']:6.1f}ms "
138
+
f"completed50={row['completed_50']:6.1f}ms count={row['count_completed']:5.2f}ms | "
139
+
f"car={row['car_mb']:6.1f}MB rss={row['rss_mb']:6.0f}MB", flush=True)
140
+
141
+
print("\n| total runs | write p50/p95 (ms) | writes/s | point | active | completed_50 | count | CAR MB | RSS MB |")
142
+
print("|---|---|---|---|---|---|---|---|---|")
143
+
for r_ in rows:
144
+
print(f"| {r_['tier']} | {r_['write_p50']:.1f} / {r_['write_p95']:.1f} "
145
+
f"| {r_['writes_per_s']:.0f} | {r_['point_lookup']:.2f} | {r_['active_by_state']:.1f} "
146
+
f"| {r_['completed_50']:.1f} | {r_['count_completed']:.2f} "
147
+
f"| {r_['car_mb']:.1f} | {r_['rss_mb']:.0f} |")
148
+
149
+
150
+
if __name__ == "__main__":
151
+
asyncio.run(main())
+11
-1
src/api.zig
+11
-1
src/api.zig
···
30
30
repo: *repo_mod.Repo,
31
31
cache: *cache_mod.Cache,
32
32
lock: std.Io.Mutex = .init,
33
+
/// commit cadence: sign+persist every N record writes. 1 = every write
34
+
/// is durably committed before its response (strictest). higher values
35
+
/// amortize commit cost at the price of a durability window — the same
36
+
/// lever as the cloud prototype's delta-flush cadence.
37
+
commit_every: usize = 1,
38
+
dirty: usize = 0,
33
39
34
40
pub fn handle(self: *Api, req: *httpz.Request, res: *httpz.Response) void {
35
41
self.dispatch(req, res) catch |err| {
···
194
200
const cb = try json_cbor.jsonToCbor(arena, parsed);
195
201
const encoded = try zat.cbor.encodeAlloc(arena, cb);
196
202
_ = try self.repo.putRecord(collection, rkey, encoded);
197
-
_ = try self.repo.commit();
203
+
self.dirty += 1;
204
+
if (self.dirty >= self.commit_every) {
205
+
_ = try self.repo.commit();
206
+
self.dirty = 0;
207
+
}
198
208
}
199
209
200
210
// ------------------------------------------------------------------
+81
-12
src/cache.zig
+81
-12
src/cache.zig
···
4
4
//! reads don't walk the MST, and it can be deleted and rebuilt from the repo
5
5
//! at any time (`hydrate` does exactly that on boot).
6
6
//!
7
-
//! backends are swappable behind the union: burner-redis (embedded,
8
-
//! in-process, the default) today; a real redis client or docket-backed
9
-
//! index slot in beside it without touching call sites.
7
+
//! backends are swappable behind the union:
8
+
//! - burner: embedded burner-redis, zero-infrastructure default
9
+
//! - redis: a real redis over TCP — the production shape (one connection
10
+
//! guarded by a mutex; pooling is the obvious next step if it shows up
11
+
//! in profiles)
10
12
11
13
const std = @import("std");
12
14
const burner = @import("burner_redis");
15
+
const redis = @import("redis");
13
16
14
17
pub const Cache = union(enum) {
15
18
burner: Burner,
19
+
redis: Redis,
16
20
17
21
pub fn initBurner(allocator: std.mem.Allocator, io: std.Io) Cache {
18
22
return .{ .burner = .{ .store = burner.Store.init(allocator, io) } };
19
23
}
20
24
25
+
pub fn initRedis(allocator: std.mem.Allocator, io: std.Io, host: []const u8, port: u16) !Cache {
26
+
return .{ .redis = .{
27
+
.io = io,
28
+
.client = try redis.Client.connect(io, allocator, host, port),
29
+
} };
30
+
}
31
+
21
32
pub fn deinit(self: *Cache) void {
22
33
switch (self.*) {
23
34
.burner => |*b| b.store.deinit(),
35
+
.redis => |*r| r.client.close(),
24
36
}
25
37
}
26
38
27
39
pub fn set(self: *Cache, key: []const u8, value: []const u8) !void {
28
40
switch (self.*) {
29
41
.burner => |*b| _ = try b.store.set(key, value, .{}),
42
+
.redis => |*r| {
43
+
r.mutex.lock(r.io) catch return error.LockFailed;
44
+
defer r.mutex.unlock(r.io);
45
+
var cmds = r.client.strings();
46
+
try cmds.set(key, value);
47
+
},
30
48
}
31
49
}
32
50
33
51
/// returned slice is owned by out_alloc
34
52
pub fn get(self: *Cache, out_alloc: std.mem.Allocator, key: []const u8) ?[]u8 {
35
-
return switch (self.*) {
36
-
.burner => |*b| b.store.get(out_alloc, key) catch null,
37
-
};
53
+
switch (self.*) {
54
+
.burner => |*b| return b.store.get(out_alloc, key) catch null,
55
+
.redis => |*r| {
56
+
r.mutex.lock(r.io) catch return null;
57
+
defer r.mutex.unlock(r.io);
58
+
var cmds = r.client.strings();
59
+
const val = cmds.get(key) catch return null;
60
+
const v = val orelse return null;
61
+
return out_alloc.dupe(u8, v) catch null;
62
+
},
63
+
}
38
64
}
39
65
40
66
// set ops back the appview's secondary indexes (runs by state, by flow)
···
43
69
pub fn sadd(self: *Cache, key: []const u8, member: []const u8) !void {
44
70
switch (self.*) {
45
71
.burner => |*b| _ = try b.store.sadd(key, &.{member}),
72
+
.redis => |*r| {
73
+
r.mutex.lock(r.io) catch return error.LockFailed;
74
+
defer r.mutex.unlock(r.io);
75
+
var cmds = r.client.sets();
76
+
_ = try cmds.sadd(key, &.{member});
77
+
},
46
78
}
47
79
}
48
80
49
81
pub fn srem(self: *Cache, key: []const u8, member: []const u8) !void {
50
82
switch (self.*) {
51
83
.burner => |*b| _ = b.store.srem(key, &.{member}) catch 0,
84
+
.redis => |*r| {
85
+
r.mutex.lock(r.io) catch return;
86
+
defer r.mutex.unlock(r.io);
87
+
var cmds = r.client.sets();
88
+
_ = cmds.srem(key, &.{member}) catch 0;
89
+
},
52
90
}
53
91
}
54
92
55
93
/// returned members owned by out_alloc
56
94
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
-
};
95
+
switch (self.*) {
96
+
.burner => |*b| return b.store.smembers(out_alloc, key) catch &.{},
97
+
.redis => |*r| {
98
+
r.mutex.lock(r.io) catch return &.{};
99
+
defer r.mutex.unlock(r.io);
100
+
var cmds = r.client.sets();
101
+
const vals = cmds.smembers(key) catch return &.{};
102
+
var out = try out_alloc.alloc([]u8, vals.len);
103
+
var n: usize = 0;
104
+
for (vals) |v| {
105
+
switch (v) {
106
+
.bulk => |b| {
107
+
const bytes = b orelse continue;
108
+
out[n] = try out_alloc.dupe(u8, bytes);
109
+
n += 1;
110
+
},
111
+
else => {},
112
+
}
113
+
}
114
+
return out[0..n];
115
+
},
116
+
}
60
117
}
61
118
62
119
pub fn scard(self: *Cache, key: []const u8) i64 {
63
-
return switch (self.*) {
64
-
.burner => |*b| b.store.scard(key) catch 0,
65
-
};
120
+
switch (self.*) {
121
+
.burner => |*b| return b.store.scard(key) catch 0,
122
+
.redis => |*r| {
123
+
r.mutex.lock(r.io) catch return 0;
124
+
defer r.mutex.unlock(r.io);
125
+
var cmds = r.client.sets();
126
+
return cmds.scard(key) catch 0;
127
+
},
128
+
}
66
129
}
67
130
};
68
131
69
132
const Burner = struct {
70
133
store: burner.Store,
134
+
};
135
+
136
+
const Redis = struct {
137
+
io: std.Io,
138
+
client: redis.Client,
139
+
mutex: std.Io.Mutex = .init,
71
140
};
72
141
73
142
test "cache set/get round trip" {
+21
-1
src/main.zig
+21
-1
src/main.zig
···
33
33
const repo = try repo_mod.Repo.boot(allocator, io, data_dir);
34
34
defer repo.deinit(allocator);
35
35
36
-
var cache = cache_mod.Cache.initBurner(allocator, io);
36
+
var cache: cache_mod.Cache = blk: {
37
+
const backend = if (std.c.getenv("PREFECT_PDS_CACHE")) |c| std.mem.span(c) else "burner";
38
+
if (std.mem.eql(u8, backend, "redis")) {
39
+
const host = if (std.c.getenv("PREFECT_PDS_REDIS_HOST")) |h| std.mem.span(h) else "127.0.0.1";
40
+
const redis_port: u16 = if (std.c.getenv("PREFECT_PDS_REDIS_PORT")) |p|
41
+
std.fmt.parseInt(u16, std.mem.span(p), 10) catch 6379
42
+
else
43
+
6379;
44
+
log.info("appview backend: redis at {s}:{d}", .{ host, redis_port });
45
+
break :blk try cache_mod.Cache.initRedis(allocator, io, host, redis_port);
46
+
}
47
+
log.info("appview backend: burner-redis (embedded)", .{});
48
+
break :blk cache_mod.Cache.initBurner(allocator, io);
49
+
};
37
50
defer cache.deinit();
38
51
52
+
const commit_every: usize = if (std.c.getenv("PREFECT_PDS_COMMIT_EVERY")) |c|
53
+
std.fmt.parseInt(usize, std.mem.span(c), 10) catch 1
54
+
else
55
+
1;
56
+
39
57
var api = api_mod.Api{
40
58
.allocator = allocator,
41
59
.io = io,
42
60
.repo = repo,
43
61
.cache = &cache,
62
+
.commit_every = commit_every,
44
63
};
64
+
if (commit_every > 1) log.info("commit cadence: every {d} writes", .{commit_every});
45
65
46
66
log.info("workspace did: {s}", .{repo.did});
47
67
log.info("repo: {d} records, data dir: {s}", .{ repo.record_count, data_dir });
+30
-8
src/repo.zig
+30
-8
src/repo.zig
···
24
24
25
25
pub const Repo = struct {
26
26
arena_state: std.heap.ArenaAllocator,
27
+
child_allocator: std.mem.Allocator,
27
28
allocator: std.mem.Allocator, // arena handle
28
29
io: std.Io,
29
30
keypair: zat.Keypair,
···
35
36
last_rev_us: u64 = 0,
36
37
data_dir: []const u8,
37
38
latest_car: ?[]const u8 = null,
39
+
latest_car_owned: bool = false,
38
40
39
41
pub fn boot(child_allocator: std.mem.Allocator, io: std.Io, data_dir: []const u8) !*Repo {
40
42
// the arena must live at its final address before anything captures
···
44
46
errdefer child_allocator.destroy(self);
45
47
self.arena_state = std.heap.ArenaAllocator.init(child_allocator);
46
48
errdefer self.arena_state.deinit();
49
+
self.child_allocator = child_allocator;
47
50
const arena = self.arena_state.allocator();
48
51
49
52
std.Io.Dir.createDirPath(.cwd(), io, data_dir) catch {};
···
61
64
self.last_rev_us = 0;
62
65
self.data_dir = try arena.dupe(u8, data_dir);
63
66
self.latest_car = null;
67
+
self.latest_car_owned = false;
64
68
65
69
try self.loadCarIfPresent();
66
70
return self;
67
71
}
68
72
69
73
pub fn deinit(self: *Repo, child_allocator: std.mem.Allocator) void {
74
+
if (self.latest_car) |old| {
75
+
if (self.latest_car_owned) child_allocator.free(old);
76
+
}
70
77
self.arena_state.deinit();
71
78
child_allocator.destroy(self);
72
79
}
···
132
139
133
140
/// sign a commit over the current tree and persist the full repo as a
134
141
/// CAR file (tmp + atomic rename). returns the commit's rev.
142
+
///
143
+
/// commit-transient memory (the serialized CAR, the block list, the
144
+
/// commit object) lives in a per-commit scratch arena — only the rev
145
+
/// survives into the repo arena, and `latest_car` is owned by the child
146
+
/// allocator and freed on replacement. without this, retaining one full
147
+
/// CAR per commit makes memory quadratic in commit count.
135
148
pub fn commit(self: *Repo) ![]const u8 {
136
149
const root = try self.tree.rootCid();
150
+
151
+
var scratch_state = std.heap.ArenaAllocator.init(self.child_allocator);
152
+
defer scratch_state.deinit();
153
+
const scratch = scratch_state.allocator();
137
154
138
155
const now_us: u64 = @intCast(@max(0, nowMicros(self.io)));
139
156
const rev_us = @max(now_us, self.last_rev_us + 1);
···
148
165
.{ .key = "prev", .value = .null },
149
166
.{ .key = "version", .value = .{ .unsigned = 3 } },
150
167
};
151
-
const unsigned_bytes = try zat.cbor.encodeAlloc(self.allocator, .{ .map = &unsigned_entries });
168
+
const unsigned_bytes = try zat.cbor.encodeAlloc(scratch, .{ .map = &unsigned_entries });
152
169
const sig = try self.keypair.sign(unsigned_bytes);
153
170
154
171
const signed_entries = unsigned_entries ++ [_]zat.cbor.Value.MapEntry{
155
-
.{ .key = "sig", .value = .{ .bytes = try self.allocator.dupe(u8, &sig.bytes) } },
172
+
.{ .key = "sig", .value = .{ .bytes = &sig.bytes } },
156
173
};
157
-
const signed_bytes = try zat.cbor.encodeAlloc(self.allocator, .{ .map = &signed_entries });
158
-
const commit_cid = try zat.cbor.Cid.forDagCbor(self.allocator, signed_bytes);
174
+
const signed_bytes = try zat.cbor.encodeAlloc(scratch, .{ .map = &signed_entries });
175
+
const commit_cid = try zat.cbor.Cid.forDagCbor(scratch, signed_bytes);
159
176
160
177
var car_blocks: std.ArrayList(zat.car.Block) = .empty;
161
-
try car_blocks.append(self.allocator, .{ .cid_raw = commit_cid.raw, .data = signed_bytes });
162
-
try self.tree.collectBlocks(&car_blocks);
178
+
try car_blocks.append(scratch, .{ .cid_raw = commit_cid.raw, .data = signed_bytes });
179
+
try self.tree.collectBlocksInto(scratch, &car_blocks);
163
180
var it = self.blocks.iterator();
164
181
while (it.next()) |e| {
165
-
try car_blocks.append(self.allocator, .{ .cid_raw = e.key_ptr.*, .data = e.value_ptr.* });
182
+
try car_blocks.append(scratch, .{ .cid_raw = e.key_ptr.*, .data = e.value_ptr.* });
166
183
}
167
184
168
-
const car_bytes = try zat.car.writeAlloc(self.allocator, .{
185
+
const car_bytes = try zat.car.writeAlloc(self.child_allocator, .{
169
186
.roots = &.{commit_cid},
170
187
.blocks = car_blocks.items,
171
188
});
189
+
errdefer self.child_allocator.free(car_bytes);
172
190
173
191
var path_buf: [512]u8 = undefined;
174
192
const path = try self.carPath(&path_buf);
···
177
195
try std.Io.Dir.writeFile(.cwd(), self.io, .{ .sub_path = tmp_path, .data = car_bytes });
178
196
try std.Io.Dir.rename(.cwd(), tmp_path, .cwd(), path, self.io);
179
197
198
+
if (self.latest_car) |old| {
199
+
if (self.latest_car_owned) self.child_allocator.free(old);
200
+
}
180
201
self.latest_car = car_bytes;
202
+
self.latest_car_owned = true;
181
203
self.commit_count += 1;
182
204
return rev;
183
205
}