Catcrafts/Crafter.Review
Watch
2
0
Fork
You've already forked Crafter.Review
 0
Code Issues 7 Pull requests Projects Releases 3 Packages Wiki Activity Actions
Crafter.Review/tests/Drivers/main.cpp

496 lines
32 KiB
C++
Raw Normal View History

feat(seatpool): the seat pool and the board several drivers share The seats of .env and the ledger, the probe, placement, settling a finished pass, the child's credentials, the OpenRouter seats, and the board every driver on the machine shares, in one shape and under one lock. A driver is alive while its lifetime lock is held; with no lock file natively its pid and start time decide, and in the sandbox it is taken for alive. :Http is the HTTP the driver needs, on FetchHTTP1: a status outside 2xx is HttpError, every other failure carries a fixed text. :Platform gains MakeDirs, with the last directory's own mode, and errno texts for its writes and locks. The tests cover the seat pool and several drivers at once, less the checks that drive the pool's Scheduler, Admission and Stacks. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-31 08:15:03 +01:00
// SPDX-License-Identifier: AGPL-3.0-only
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
// Several drivers at once: the board (seats, memory, the cap), one full run
// per hash, one driver per scratch dir, the merged unit-state, the locked
// ledger, the shared memory record. The checks that run the Scheduler,
// Admission, Stacks, the stack keys and the run's state store come with the
// pool and the driver.
//
// Another driver is a live process of its own (`sleep 300`), so its entries
// stand until it dies; two drivers recording into one memory file are two
// runs of this binary (`--record`).
#include <signal.h>
#include <stdlib.h>
#include <unistd.h>
import std;
import Crafter.Review;
namespace fs = std::filesystem;
using namespace Crafter::Review;
using namespace std::chrono_literals;
namespace {
std::int32_t Failures = 0;
void Check(std::string_view name, bool cond, std::string_view detail = {}) {
std::println("{}{}{}", cond ? " ok " : " FAIL ", name, !cond && !detail.empty() ? std::format(" ({})", detail) : std::string());
if (!cond) ++Failures;
}
void Require(bool cond, std::string_view what) {
if (!cond) throw std::runtime_error(std::string(what));
}
Json With(Json value, const Json& changes) {
for (auto it = changes.begin(); it != changes.end(); ++it) value[it.key()] = it.value();
return value;
}
void Write(const fs::path& path, std::string_view text) {
fs::create_directories(path.parent_path());
std::ofstream(path, std::ios::binary) << text;
}
// The board file, as JSON.
Json Read(const fs::path& path) {
return Dyn::LoadJsonFile(path);
}
std::set<std::string> Keys(const Json& object) {
std::set<std::string> keys;
for (auto it = object.begin(); it != object.end(); ++it) keys.insert(it.key());
return keys;
}
bool AnyContains(const std::vector<std::string>& lines, std::string_view part) {
return std::ranges::any_of(lines, [&](const std::string& line) { return line.contains(part); });
}
Json& At(std::vector<Json>& seats, std::string_view id) {
for (Json& s : seats) {
if (s.at("id") == id) return s;
}
throw std::out_of_range(std::string(id));
}
// A driver of its own: a live process whose pid and start time name it.
struct Sleeper {
std::int64_t pid = Process::Start({.argv = {"sleep", "300"}}).pid;
std::string key = std::format("{}:{}", pid, Platform::ProcStart(pid).value_or(0));
void Kill() const {
Process::Signal(pid, SIGKILL);
Process::Wait(pid);
}
};
std::int32_t Run() {
std::println("several drivers at once: the board (seats, memory, stacks, the cap), one full run per hash, one driver per scratch dir, the merged unit-state, the locked ledger");
Platform::SetMode(Platform::Mode::Native);
const fs::path td = fs::temp_directory_path() / std::format("crafter-review-drivers-{}", Platform::PublishedPid());
fs::remove_all(td);
fs::create_directories(td);
const fs::path path = td / "seat-board.json";
// The lifetime locks ClaimScratch looks for are beside the board.
setenv("REVIEW_GATES_BOARD", path.c_str(), 1);
std::mutex logLock;
std::vector<std::string> logs;
auto log = [&](const std::string& line) {
const std::lock_guard guard(logLock);
logs.push_back(line);
};
const double now = Platform::Now();
const std::int64_t start = Platform::ProcStart(getpid()).value_or(0);
const std::string mine = std::format("{}:{}", getpid(), start);
SeatPool::Board A(path, {}, std::nullopt, log);
Check("identity: pid + start time from /proc, alive; a nonexistent pid, a recycled pid (another start time) and another boot are not",
A.Me() == mine && A.Alive(A.Me(), Json::object()) && !A.Alive("2147483647:1", Json::object()) && !A.Alive(std::format("{}:{}", getpid(), start + 1), Json::object())
&& !SeatPool::Board(path, "x:y", "bootA").Alive(mine, {{mine, {{"boot", "bootB"}}}}));
Check("seat keys: one token under two .env numbers is one account, the key carries no secret, the login is local, the kinds differ",
SeatPool::NewSeat("1", "oauth", "tok-one").at("boardKey") == SeatPool::NewSeat("5", "oauth", "tok-one").at("boardKey") && !SeatPool::SeatKey("oauth", "tok-one").contains("tok-one")
&& SeatPool::SeatKey("local", "") == "local" && SeatPool::SeatKey("api", "k") != SeatPool::SeatKey("oauth", "k"));
const Json cfg = SeatPool::PoolConfig({});
const Json acfg = SeatPool::ApiConfig({{"REVIEW_GATES_API_CONCURRENCY", "8"}, {"REVIEW_GATES_API_LENS_COST_USD", "1"}});
auto mk = [&](const std::string& id, const std::string& kind, const std::string& secret, std::optional<std::int64_t> fh = std::nullopt, std::optional<std::int64_t> sd = std::nullopt,
const std::string& state = "ok") {
Json s = SeatPool::NewSeat(id, kind, secret);
s["state"] = state;
if (fh) {
s["windows"] = {{"five_hour", {{"utilization", *fh}, {"resetsAt", now + 3600}}}, {"seven_day", {{"utilization", sd.value_or(0)}, {"resetsAt", now + 86400}}}};
const auto [util, binding] = Seats::SeatUtil(s.at("windows"));
s["util"] = util;
s["binding"] = binding;
}
return s;
};
// The jobs the choosers hand out: they must outlive the place that
// registers them.
std::deque<Json> jobs;
std::mutex jobsLock;
auto chooserFor = [&](const std::string& id, const std::string& label, const Json& extra = Json::object()) {
return SeatPool::Board::Chooser([&jobs, &jobsLock, id, label, extra](const std::vector<Json*>& ranked, const SeatPool::Others&) -> std::optional<std::pair<Json*, Json*>> {
for (Json* s : ranked) {
if (s->at("id") == id) {
const std::lock_guard guard(jobsLock);
jobs.push_back(With({{"label", label}, {"timeoutMin", 20}}, extra));
return std::pair(s, &jobs.back());
}
}
return std::nullopt;
});
};
// A call on [seat]: the seat comes back changed.
auto fold1 = [&](SeatPool::Board& board, Json& seat, double at) {
std::vector<Json> one = {seat};
std::vector<std::string> lines = board.Fold(one, at);
seat = one.front();
return lines;
};
auto place1 = [&](SeatPool::Board& board, Json& seat, const Json& c, const SeatPool::Board::Chooser& chooser) {
std::vector<Json> one = {seat};
const bool picked = board.Place(one, now, c, acfg, chooser).picked.has_value();
seat = one.front();
return picked;
};
const Sleeper child;
SeatPool::Board B(path, child.key, std::nullopt, log);
A.Join({{"run", "A"}}, now);
B.Join({{"run", "B"}}, now);
std::vector<Json> sa = {mk("1", "oauth", "t1", 10, 50), mk("2", "oauth", "t2", 10, 50), mk("api1", "api", "k1")};
std::vector<Json> sb = {mk("1", "oauth", "t1", 10, 50), mk("2", "oauth", "t2", 10, 50), mk("api1", "api", "k1")};
for (std::int32_t i = 0; i < 3; ++i) {
const SeatPool::Board::Placed placed = B.Place(sb, now, cfg, acfg, chooserFor("1", std::format("b{}", i)));
Require(placed.picked && placed.picked->first->at("id") == "1", "B places on seat 1");
}
for (std::int32_t i = 0; i < 2; ++i) {
const SeatPool::Board::Placed placed = B.Place(sb, now, cfg, acfg, chooserFor("api1", std::format("ba{}", i)));
Require(placed.picked && placed.picked->first->at("id") == "api1", "B places on api1");
}
// The pool width is the machine's cap: B's five passes count against
// A's.
SeatPool::Board::Placed placed = A.Place(sa, now, cfg, acfg, chooserFor("2", "cap0"), 5);
Check("the cap counts every driver: five in flight on the machine hold a sixth at cap 5, the seats still ranked", !placed.picked && placed.full == 5 && !placed.ranked.empty());
placed = A.Place(sa, now, cfg, acfg, chooserFor("2", "cap0"), 6);
Check("...and a sixth goes at cap 6; it counts from then on", placed.picked && placed.full == 0 && SeatPool::Board::TotalInflight(Read(path)) == 6);
A.Finish(At(sa, "2"), "cap0", now);
A.Fold(sa, now);
const Json f1 = At(sa, "1").at("foreign");
const Json fa = At(sa, "api1").at("foreign");
Check("three passes of B on seat 1 and two on api1 fold into A as foreign, at B's own per-pass figures",
f1.at("count") == 3 && Dyn::Equal(f1.at("cost5h"), 0.75) && Dyn::Equal(f1.at("costUsd"), 0) && fa.at("count") == 2 && Dyn::Equal(fa.at("costUsd"), 2.0)
&& Dyn::Equal(f1.at("drivers"), {{B.Me(), 3}}),
std::format("{} {}", DumpSpaced(f1), DumpSpaced(fa)));
Json s84 = mk("x", "oauth", "tx", 84, 50);
s84["foreign"] = {{"count", 3}, {"cost5h", 1.5}, {"costUsd", 0}};
const bool over = !SeatPool::Eligible(s84, now, cfg, acfg);
Json s84Bare = With(s84, {{"foreign", {{"count", 0}, {"cost5h", 0.0}, {"costUsd", 0.0}}}});
Check("the 5-hour projection counts the foreign passes: 84 + 1.5 + 0.25 is over 85, without them under", over && SeatPool::Eligible(s84Bare, now, cfg, acfg));
Json a8 = mk("api9", "api", "k9");
a8["foreign"] = {{"count", 6}, {"cost5h", 0}, {"costUsd", 6.0}};
a8["running"] = 2;
Json a8One = With(a8, {{"running", 1}});
Check("an API key's 8 slots count every driver: 6 foreign + 2 running is full, 6 + 1 is not", !SeatPool::Eligible(a8, now, cfg, acfg) && SeatPool::Eligible(a8One, now, cfg, acfg));
Check("...and its budget the other drivers' reserve: $6 foreign + $1 running + $1 for the next is over a $7 budget, inside $8",
!SeatPool::Eligible(a8One, now, cfg, With(acfg, {{"budgetUsd", 7}})) && SeatPool::Eligible(a8One, now, cfg, With(acfg, {{"budgetUsd", 8}})));
std::vector<Json> es = {mk("e1", "oauth", "te1", 20, 10), mk("e2", "oauth", "te2", 20, 10)};
es[0]["foreign"] = {{"count", 10}, {"cost5h", 5.0}, {"costUsd", 0}};
std::vector<std::string> order;
for (const Json* s : SeatPool::Ranked(es, now, cfg, With(acfg, {{"budgetUsd", 0}}))) order.push_back(s->at("id").get<std::string>());
Check("ranked: equal readings, the seat with foreign load ranks second", order == std::vector<std::string>{"e2", "e1"}, Dyn::Join(order, ","));
// Readings merge by max per reset epoch, newer epoch wins, expired
// ignored.
Json w = {{"five_hour", {{"utilization", 40}, {"resetsAt", now + 3600}}}};
const std::vector<std::string> ch = SeatPool::MergeWindows(w, {{"five_hour", {{"utilization", 44}, {"resetsAt", now + 3600}}}}, now);
const bool higher = ch == std::vector<std::string>{"five_hour"} && w["five_hour"]["utilization"] == 44;
Check("windows merge: the same epoch takes the higher utilization, a lower one changes nothing",
higher && SeatPool::MergeWindows(w, {{"five_hour", {{"utilization", 30}, {"resetsAt", now + 3600}}}}, now).empty() && w["five_hour"]["utilization"] == 44);
SeatPool::MergeWindows(w, {{"five_hour", {{"utilization", 3}, {"resetsAt", now + 7200}}}}, now);
Check("...a newer epoch replaces an older one, however low", Dyn::Equal(w["five_hour"], {{"utilization", 3}, {"resetsAt", now + 7200}}));
Json expired = {{"five_hour", {{"utilization", 0}, {"resetsAt", 0}}}};
Check("...a reading whose reset has passed is ignored on either side",
SeatPool::MergeWindows(w, {{"five_hour", {{"utilization", 90}, {"resetsAt", now - 100}}}}, now).empty() && w["five_hour"]["utilization"] == 3
&& SeatPool::MergeWindows(expired, {{"five_hour", {{"utilization", 50}, {"resetsAt", now + 100}}}}, now) == std::vector<std::string>{"five_hour"});
Json sx = mk("1", "oauth", "t1");
sx["windows"] = {{"five_hour", {{"utilization", 0}, {"resetsAt", 0}}}};
A.Publish({sx}, now, true);
Check("...and a locally expired window (expire_windows) is never published",
!Dyn::Has(Dyn::Or(Dyn::Get(Read(path).at("seats").at(sx.at("boardKey").get<std::string>()), "windows"), Dyn::EmptyDict()), "five_hour"));
At(sb, "1")["windows"]["five_hour"]["utilization"] = 40;
B.Publish({At(sb, "1")}, now);
fold1(A, At(sa, "1"), now);
Check("a higher reading of the same epoch on the board reaches the other driver's seat, util and binding recomputed",
At(sa, "1")["windows"]["five_hour"]["utilization"] == 40 && At(sa, "1")["util"] == 50 && At(sa, "1")["binding"] == "seven_day");
const std::optional<std::string> before = Platform::ReadFile(path);
SeatPool::Board C(path, A.Me(), std::nullopt, log);
C.Publish({At(sa, "1")}, now);
fold1(C, At(sa, "1"), now);
Check("publish then fold of what the board already holds writes nothing (the merge is idempotent)", Platform::ReadFile(path) == before);
// Verdicts: an OAuth seat's out stands until its window resets,
// invalid while its driver lives; never a paid seat's.
Json weekly = {{"status", "limit"},
{"reason", "weekly"},
{"rateType", "seven_day"},
{"rateLimit", {{"five_hour", {{"utilization", 12}, {"resetsAt", now + 3600}}}, {"seven_day", {{"utilization", 100}, {"resetsAt", now + 500}}}}}};
const bool requeued = SeatPool::Settle(At(sb, "2"), weekly, now, cfg, acfg, nullptr, log);
B.Finish(At(sb, "2"), "none", now);
A.Fold(sa, now);
const std::string key2 = At(sa, "2").at("boardKey").get<std::string>();
Check("an OAuth seat another driver found out (seven-day rejected) is out here, until that window resets",
requeued && At(sa, "2")["state"] == "out" && Dyn::Str(At(sa, "2")["error"]).contains("seen by pid") && Read(path)["seats"][key2]["state"]["until"].get<double>() == now + 530,
Dyn::Str(At(sa, "2")["error"]));
Json sa2 = mk("2", "oauth", "t2", 10, 50);
fold1(A, sa2, now + 600);
const Json record2 = Dyn::Get(Read(path).at("seats"), key2);
Check("...and not once that reset has passed (the record is swept)", sa2["state"] == "ok" && Dyn::Get(Dyn::Or(record2, Dyn::EmptyDict()), "state").is_null());
sa2 = mk("2", "oauth", "t2", 10, 50, "invalid");
sa2["error"] = "mine";
fold1(A, sa2, now);
Check("a local out/invalid is not overwritten by the board's", sa2["state"] == "invalid" && sa2["error"] == "mine");
sa2 = mk("2", "oauth", "t2", 10, 50);
A.Publish({sa2}, now, true);
Check("a seat our own probe reads ok clears the verdict another driver left", Read(path)["seats"][key2]["state"].is_null());
At(sb, "1")["state"] = "invalid";
At(sb, "1")["error"] = "authentication failure (401)";
B.Publish({At(sb, "1")}, now);
Json sa1 = mk("1", "oauth", "t1", 10, 50);
fold1(A, sa1, now);
Check("an invalid token seen by another driver is adopted while that driver lives", sa1["state"] == "invalid" && Dyn::Str(sa1["error"]).contains("authentication"));
const std::string kApi = At(sa, "api1").at("boardKey").get<std::string>();
Json bd = Read(path);
if (!bd["seats"].contains(kApi)) bd["seats"][kApi] = {{"id", "api1"}, {"kind", "api"}, {"windows", Json::object()}, {"pauseUntil", 0}, {"state", nullptr}, {"inflight", Json::object()}};
bd["seats"][kApi]["state"] = {{"state", "out"}, {"error", "budget"}, {"until", now + 9999}, {"by", B.Me()}};
bd["seats"][kApi]["pauseUntil"] = now + 90;
Write(path, DumpSpaced(bd));
Json saApi = mk("api1", "api", "k1");
fold1(A, saApi, now);
Check("a paid seat's verdict never crosses drivers (the ledger carries a credit-out, a budget-out is each driver's own) - its pause does, until the same moment",
saApi["state"] == "ok" && saApi["pauseUntil"].get<double>() == now + 90 && !SeatPool::Eligible(saApi, now, cfg, acfg) && SeatPool::Eligible(saApi, now + 91, cfg, acfg));
child.Kill();
sa1 = mk("1", "oauth", "t1", 10, 50);
const std::vector<std::string> lines = fold1(A, sa1, now);
Check("a driver that died is swept with its in-flight entries and its invalid verdict, and the sweep says so",
sa1["state"] == "ok" && sa1["foreign"]["count"] == 0 && !Read(path).at("drivers").contains(B.Me()) && AnyContains(lines, "(dead)"), Dyn::Join(lines, "; "));
SeatPool::Board Z(path, std::format("{}:{}", getpid(), start + 1), std::nullopt, log);
Z.Join({{"run", "Z"}}, now);
Json z0 = mk("1", "oauth", "t1", 10, 50);
place1(Z, z0, cfg, chooserFor("1", "z0"));
Json sz = mk("1", "oauth", "t1", 10, 50);
fold1(A, sz, now);
Check("a driver record with a live pid but another start time is a recycled pid: swept", sz["foreign"]["count"] == 0 && !Read(path).at("drivers").contains(Z.Me()));
Json data = {{"drivers", {{A.Me(), Json::object()}}},
{"seats",
{{"k",
{{"id", "1"},
{"kind", "oauth"},
{"windows", Json::object()},
{"pauseUntil", 0},
{"state", nullptr},
{"inflight", {{std::format("{}/late", A.Me()), {{"since", now - 9999}, {"deadline", now - 1}}}, {std::format("{}/live", A.Me()), {{"since", now}, {"deadline", now + 100}}}}}}}}}};
const auto [swept, sweptLines] = A.Sweep(data, now);
A.Finish(mk("1", "oauth", "t1", 10, 50), "late", now);
Check("an in-flight entry past its deadline is swept, a live one kept; finishing a pass the sweep already took is harmless",
swept && Keys(data["seats"]["k"]["inflight"]) == std::set<std::string>{std::format("{}/live", A.Me())} && AnyContains(sweptLines, "past their deadline"), Dyn::Join(sweptLines, "; "));
// A second live driver: leave, the atomic place, its own projection.
const Sleeper child2;
SeatPool::Board D(path, child2.key, std::nullopt, log);
D.Join({{"run", "D"}}, now);
Json d0 = mk("1", "oauth", "t1", 10, 50);
place1(D, d0, cfg, chooserFor("1", "d0"));
D.Leave();
D.Leave();
Json sl = mk("1", "oauth", "t1", 10, 50);
fold1(A, sl, now);
Check("leave takes the driver and its in-flight entries off the board; a second leave is harmless", sl["foreign"]["count"] == 0 && !Read(path).at("drivers").contains(D.Me()));
D.Join({{"run", "D"}}, now);
const Json cfg2 = With(cfg, {{"lensCostPct", 2}, {"maxUtil", 85}});
// An account of its own: its finish publishes 82.
Json sa82 = mk("4", "oauth", "t4", 82, 10);
Json sd82 = mk("4", "oauth", "t4", 82, 10);
std::mutex resultsLock;
std::vector<std::pair<bool, bool>> results;
auto race = [&](SeatPool::Board& board, Json& seat) {
std::vector<Json> one = {seat};
Json job = {{"label", "race"}, {"timeoutMin", 20}};
const SeatPool::Board::Placed got = board.Place(one, now, cfg2, acfg, [&job](const std::vector<Json*>& ranked, const SeatPool::Others&) -> std::optional<std::pair<Json*, Json*>> {
std::this_thread::sleep_for(50ms);
if (ranked.empty()) return std::nullopt;
return std::pair(ranked.front(), &job);
});
seat = one.front();
const std::lock_guard guard(resultsLock);
results.emplace_back(!got.ranked.empty(), got.picked.has_value());
};
{
const std::jthread ta(race, std::ref(A), std::ref(sa82));
const std::jthread tb(race, std::ref(D), std::ref(sd82));
}
const auto launched = std::ranges::count_if(results, [](const auto& r) { return r.second; });
const auto roomless = std::ranges::count_if(results, [](const auto& r) { return !r.first; });
Check("place is one locked section: two drivers racing for a seat with room for one pass get exactly one launch",
launched == 1 && roomless == 1 && Read(path)["seats"][sa82.at("boardKey").get<std::string>()]["inflight"].size() == 1, std::format("{} launched, {} without room", launched, roomless));
A.Finish(sa82, "race", now);
D.Finish(sd82, "race", now);
Json overCeiling = mk("o", "oauth", "to", 90, 10);
overCeiling["windows"]["five_hour"]["resetsAt"] = now + 600;
overCeiling["foreign"] = {{"count", 1}, {"cost5h", 0.5}, {"costUsd", 0}};
const Json overBare = With(overCeiling, {{"foreign", {{"count", 0}, {"cost5h", 0.0}, {"costUsd", 0.0}}}});
Check("earliest_resume: with another driver's pass in flight the wait is one re-check, else the window reset",
SeatPool::EarliestResume({overCeiling}, now, 30) == now + 30 && SeatPool::EarliestResume({overCeiling}, now) == now + 630 && SeatPool::EarliestResume({overBare}, now, 30) == now + 630);
D.Leave();
D.Join({{"run", "D"}}, now);
Json sd1 = mk("1", "oauth", "t1", 10, 50);
place1(D, sd1, cfg2, chooserFor("1", "m0"));
place1(D, sd1, cfg2, chooserFor("1", "m1"));
Json sm = mk("1", "oauth", "t1", 10, 50);
fold1(A, sm, now);
// The board's reading of the account (40, B's) folded in before the
// projection.
const Json base = sm["windows"]["five_hour"]["utilization"];
Check("a driver's passes fold at its own projection (2 % each) into another driver's (0.25 %)",
Dyn::Equal(sm["foreign"], {{"count", 2}, {"cost5h", 4.0}, {"costUsd", 0}, {"drivers", {{D.Me(), 2}}}}) && base == 40 && SeatPool::OauthProjected(sm, cfg) == base.get<double>() + 4.0,
std::format("{} {}", DumpSpaced(sm["foreign"]), base.dump()));
const std::optional<std::string> unviewed = Platform::ReadFile(path);
Json v = A.View(now);
const std::string km = sm.at("boardKey").get<std::string>();
Check("view: the drivers and, per account, the passes in flight by driver - nothing written",
v["drivers"].contains(A.Me()) && v["drivers"].contains(D.Me()) && v["seats"][km]["inflight"] == 2 && Dyn::Equal(v["seats"][km]["by"], {{D.Me(), 2}}) && v["drivers"][D.Me()]["inflight"] == 2
&& Platform::ReadFile(path) == unviewed);
D.Leave();
Check("no secret on the board", !DumpSpaced(Read(path).at("seats")).contains("t1") && !Platform::ReadFile(path).value_or("").contains("k1"));
// Memory across drivers (2026-10-31): what the board's passes were
// admitted at and the stacks they hold.
D.Join({{"run", "D"}}, now);
Json sE = mk("1", "oauth", "t1", 10, 50);
place1(D, sE, cfg, chooserFor("1", "e0", {{"_admitted", 840.0}, {"stackKey", "http://127.0.0.1:5173"}}));
place1(D, sE, cfg, chooserFor("1", "e1", {{"_admitted", 210.0}}));
// Another account: memory and stacks are the machine's, not a seat's.
Json sA = mk("5", "oauth", "t5", 10, 50);
fold1(A, sA, now);
Json entry = Read(path)["seats"][sE.at("boardKey").get<std::string>()]["inflight"][std::format("{}/e0", D.Me())];
const SeatPool::Others& others = A.OthersNow();
Check("a pass on the board carries the memory it was admitted at and the stack it holds; another driver's fold sees every pass on the machine, whatever account",
Dyn::Equal(entry["mb"], 840.0) && entry["stack"] == "http://127.0.0.1:5173" && others.count == 2 && others.mb == 1050.0 && others.stacks == std::set<std::string>{"http://127.0.0.1:5173"}
&& others.unmeasured == 0 && sA["foreign"]["count"] == 0,
std::format("{} passes, {} MB", others.count, others.mb));
// The memory base the gates budget from.
D.Leave();
const auto r0 = A.MemoryBase(50'000, now);
D.Join({{"run", "D"}}, now);
place1(D, sE, cfg, chooserFor("1", "f0", {{"_admitted", 840.0}}));
const auto r1 = A.MemoryBase(30'000, now);
// A pass that records no memory (a driver that does not measure it).
place1(D, sE, cfg, chooserFor("1", "f1"));
const auto r2 = A.MemoryBase(30'000, now);
Check("memory base: with no pass in flight on the machine the reading of now, recorded; with other drivers' measured passes in flight the board's reading from before them; with one that records none, the reading of now",
r0 == std::pair<double, std::string>{50'000, "no pass in flight on this machine"} && r1.first == 50'000 && r1.second.contains("the board's reading") && r1.second.contains("840 MB")
&& r2.first == 30'000 && r2.second.contains("record no memory"),
std::format("{} | {} | {}", r0.second, r1.second, r2.second));
D.Leave();
A.Leave();
child2.Kill();
// One live driver per scratch dir.
const fs::path sd = td / "h8-run-2";
const Sleeper child4;
fs::create_directories(sd);
Write(sd / ".driver", child4.key);
const bool held = SeatPool::ClaimScratch(sd);
child4.Kill();
const bool takenOver = SeatPool::ClaimScratch(sd);
Check("a scratch dir a live driver claimed is refused; a dead claim is taken over; a fresh dir is claimed with our pid",
!held && takenOver && Platform::ReadFile(sd / ".driver") == mine && SeatPool::ClaimScratch(td / "h8-run-3") && SeatPool::ClaimScratch(sd));
// One full run per hash.
const fs::path gd = td / "gate-state";
fs::create_directories(gd);
auto f = std::make_unique<State::RunLock>(gd, "commits", "abcd1234");
try {
const State::RunLock second(gd, "commits", "abcd1234");
Check("run lock: a second full run on the hash is refused", false);
} catch (const State::RunLocked& e) {
Check("run lock: a second full run on the hash is refused, the holder not yet identified", std::string_view(e.what()).contains("not identified"), e.what());
}
f->WriteInfo({{"pid", 1}, {"since", "s"}, {"run_n", 1}, {"scratch", "x"}});
f->WriteInfo({{"pid", 1}, {"since", "s"}, {"run_n", 2}, {"scratch", "x"}});
try {
const State::RunLock second(gd, "commits", "abcd1234");
Check("run lock: ...named once it wrote itself", false);
} catch (const State::RunLocked& e) {
const std::optional<Json> holder = State::LockHolder(gd, "commits", "abcd1234");
Check("run lock: ...named once it wrote itself (one JSON document after two writes)",
std::string_view(e.what()).contains("pid 1") && std::string_view(e.what()).contains("run 2") && holder && holder->at("run_n") == 2, e.what());
}
auto takes = [&](std::string_view hash8) {
try {
const State::RunLock taken(gd, "commits", hash8);
return true;
} catch (const State::RunLocked&) {
return false;
}
};
Check("run lock: another hash is free; nobody holds an untaken one", takes("ffff0000") && !State::LockHolder(gd, "commits", "00000000"));
f.reset();
Check("run lock: released on close (the kernel releases it on death the same way)", !State::LockHolder(gd, "commits", "abcd1234") && takes("abcd1234"));
// unit-state merge-on-write
const fs::path sp = td / "unit-state.json";
const Json a1 = {{"unit", "X"}, {"hash", "H1"}};
const Json b1 = {{"unit", "Y"}, {"hash", "H1"}};
const Json read1 = {{"hash", "H1"}, {"round", 1}, {"run", "r1"}};
State::SaveState(sp, {{"ka", a1}}, {}, read1);
Json m = State::SaveState(sp, {{"kb", b1}}, {}, read1);
Check("unit-state merge: two drivers' stores both survive, each with its read",
Keys(m["entries"]) == std::set<std::string>{"ka", "kb"} && Dyn::Equal(m["entries"]["ka"]["reads"], Json::array({read1})) && Keys(State::LoadState(sp)["entries"]) == std::set<std::string>{"ka", "kb"});
const std::vector<std::string> dropKa = {"ka"};
m = State::SaveState(sp, {{"ka", a1}}, dropKa);
Check("...a drop wins over a set in the same store", Keys(m["entries"]) == std::set<std::string>{"kb"});
const Json read2 = {{"hash", "H2"}, {"round", 1}, {"run", "r2"}};
State::SaveState(sp, {{"kb", b1}}, {}, read2);
m = State::SaveState(sp, {{"kb", b1}}, {}, read2);
std::vector<std::string> reads;
for (const Json& x : m["entries"]["kb"]["reads"]) reads.push_back(std::format("{}/{}", x.at("hash").get<std::string>(), x.at("round").dump()));
Check("...a read of another hash is a second read, counted once however often it is stored", reads == std::vector<std::string>{"H1/1", "H2/1"}, Dyn::Join(reads, " "));
Check("...a whole pass carries no count", !State::SaveState(sp, {{"kw", {{"kind", "whole"}}}}, {}, {{"hash", "H"}, {"round", 1}, {"run", "r"}})["entries"]["kw"].contains("reads"));
Write(sp, "garbage");
Check("...garbage on disk is the empty state", Dyn::Equal(State::SaveState(sp, {{"k", {{"unit", "Z"}}}})["entries"], {{"k", {{"unit", "Z"}}}}));
const Json old = {{"version", 1}, {"entries", {{"q", {{"unit", "Q"}}}}}, {"personaPass", nullptr}};
State::SaveWhole(sp, old);
Check("...sets=None is the whole-file write as before", Dyn::Equal(Dyn::LoadJsonFile(sp), old));
// The ledger: two drivers on one file, every write a locked delta.
const fs::path lp = td / "seat-ledger.json";
SeatPool::Ledger la = SeatPool::Ledger::Load(lp, acfg);
SeatPool::Ledger lb = SeatPool::Ledger::Load(lp, acfg);
la.Add("api9", 2.0, now);
lb.Add("api9", 3.0, now);
la.Add("api9", 2.0, now);
lb.MarkOut("api9", "credit balance too low", now);
const Json last = Dyn::LoadJsonFile(lp).at("api9");
Check("two drivers on one ledger file: every add and the out mark survive, the last writer no longer wins",
Dyn::Equal(last.at("spentUsd"), 7.0) && last.at("procs") == 3 && Dyn::Truthy(last.at("outSince")) && Dyn::Equal(la.Entry("api9", now).at("spentUsd"), 7.0), DumpSpaced(last));
SeatPool::Ledger lc(lp, acfg, {{"api8", {{"cycle", "2099-01-01"}, {"spentUsd", 5.0}, {"procs", 1}, {"outSince", nullptr}, {"outReason", nullptr}}}});
Check("...an entry of a later cycle (another driver's cycle day) is not wiped", Dyn::Equal(lc.Entry("api8", now).at("spentUsd"), 5.0));
// The memory record: every driver's peaks in one file.
const fs::path mp = td / "memory.json";
std::vector<std::int64_t> recorders;
for (const std::string_view environment : {"none", "browser"}) {
recorders.push_back(Process::Start({.argv = {Platform::SelfExe().string(), "--record", mp.string(), td.string(), std::string(environment)}}).pid);
}
for (const std::int64_t pid : recorders) Process::Wait(pid);
const Memory::Calibration mc(mp, td, {});
Check("two drivers recording peaks into one memory file at once: every peak of both is kept", mc.Peaks("none").size() == 20 && mc.Peaks("browser").size() == 20,
std::format("{} {}", mc.Peaks("none").size(), mc.Peaks("browser").size()));
fs::remove_all(td);
return Failures == 0 ? 0 : 1;
}
}
int main(int argc, char** argv) {
// A second driver recording its peaks: twenty passes on one environment.
if (argc == 5 && std::string_view(argv[1]) == "--record") {
Memory::Calibration calibration(argv[2], argv[3], {});
for (std::int32_t i = 0; i < 20; ++i) calibration.Record(argv[4], 100.0 + i);
return 0;
}
try {
return Run();
} catch (const std::exception& error) {
std::println(std::cerr, "drivers: {}", error.what());
return 1;
}
}
Reference in a new issue Copy permalink