Add persistent AFK reserves for both engines (Refs #12)
This commit is contained in:
@@ -0,0 +1,158 @@
|
||||
#include "afk.hpp"
|
||||
#include <json.hpp>
|
||||
#include <cassert>
|
||||
#include <fstream>
|
||||
#include <iostream>
|
||||
using utils::afk::journal;
|
||||
using utils::afk::parse_acknowledgement;
|
||||
using namespace std::chrono_literals;
|
||||
namespace fs = std::filesystem;
|
||||
const std::string server = "test-server", endpoint = "https://example.test/api/afk";
|
||||
std::string reserve(const std::string& id)
|
||||
{
|
||||
return nlohmann::json{{"server",server},{"guid","18446744073709551615"},{"match","match"},
|
||||
{"request_id",id+"-reserve"},{"action","reserve"},{"session_id",id},{"seconds",600}}.dump();
|
||||
}
|
||||
std::string request(const std::string& id, const std::string& action)
|
||||
{
|
||||
return nlohmann::json{{"server",server},{"guid","18446744073709551615"},{"match","match"},
|
||||
{"request_id",id},{"action",action},{"session_id","pending"},{"confirmed",true},{"packs",1},
|
||||
{"seq",1},{"active_seconds",0},{"kills",0},{"headshots",0},{"rounds",0},{"revives",0},{"activity","playing"}}.dump();
|
||||
}
|
||||
template <typename F> void must_fail(F fn)
|
||||
{
|
||||
bool failed = false;
|
||||
try { fn(); } catch (const std::exception&) { failed = true; }
|
||||
assert(failed);
|
||||
}
|
||||
int main(int argc, char** argv)
|
||||
{
|
||||
assert(argc >= 2);
|
||||
const fs::path root(argv[1]);
|
||||
const journal queue(root,server,endpoint);
|
||||
if (argc == 3)
|
||||
{
|
||||
const std::string mode(argv[2]);
|
||||
if (mode == "produce")
|
||||
{
|
||||
queue.prepare(reserve("pending"));
|
||||
queue.store(request("pending-start","start"));
|
||||
queue.prepare(reserve("authorized")); queue.authorize("authorized",120);
|
||||
queue.prepare(reserve("active")); queue.authorize("active",120);
|
||||
const auto now = journal::clock::now();
|
||||
queue.begin("active",now); assert(queue.checkpoint("active",now+31s) == 31);
|
||||
queue.store(request("purchase","purchase")); queue.store(request("event","event"));
|
||||
}
|
||||
else if (mode == "recover")
|
||||
{
|
||||
// Separate process: the same match ID cannot recreate a live clock.
|
||||
assert(queue.remaining("active") == -1);
|
||||
must_fail([&]{queue.begin("active");});
|
||||
queue.recover("match");
|
||||
const auto values = queue.load();
|
||||
assert(values.size() == 5);
|
||||
int recovered = 0;
|
||||
for (const auto& value : values)
|
||||
{
|
||||
if (value.action != "settle") continue;
|
||||
auto body = nlohmann::json::parse(value.body);
|
||||
assert(value.id.size() == 25);
|
||||
if (body.at("session_id") == "active")
|
||||
{ assert(body.at("consumed_seconds").is_null()); assert(body.at("end_reason") == "crash"); assert(body.at("confirmed_seconds") == 31); }
|
||||
else { assert(body.at("consumed_seconds") == 0); assert(body.at("end_reason") == "never_started"); }
|
||||
++recovered;
|
||||
}
|
||||
assert(recovered == 3);
|
||||
queue.recover("new-match"); // Stable requests are unchanged across restarts.
|
||||
assert(queue.load().size() == values.size());
|
||||
for (size_t i=0; i<values.size(); ++i) assert(queue.load()[i].body == values[i].body);
|
||||
for (const auto& value : values) queue.erase(value.id);
|
||||
}
|
||||
else assert(mode == "empty" && queue.load().empty());
|
||||
return 0;
|
||||
}
|
||||
fs::create_directories(root);
|
||||
queue.store(request("one","purchase")); queue.store(request("one","purchase"));
|
||||
assert(queue.load().size() == 1);
|
||||
must_fail([&]{queue.store(request("one","event"));});
|
||||
auto changed = nlohmann::json::parse(request("one","purchase")); changed["packs"] = 2;
|
||||
must_fail([&]{queue.store(changed.dump());});
|
||||
const journal other_server(root,"another-server",endpoint), other_endpoint(root,server,endpoint+"/another");
|
||||
assert(other_server.load().empty() && other_endpoint.load().empty());
|
||||
queue.erase("one");
|
||||
queue.store(request("first-event","event"));
|
||||
auto snapshot = nlohmann::json::parse(request("second-event","event"));
|
||||
snapshot["seq"] = 2; snapshot["revives"] = 2;
|
||||
queue.store(snapshot.dump());
|
||||
assert(queue.load().size() == 1 && queue.load()[0].id == "second-event");
|
||||
queue.erase("first-event"); // An in-flight older ACK cannot delete the new snapshot.
|
||||
assert(queue.load().size() == 1);
|
||||
auto regressed = snapshot; regressed["request_id"]="third-event"; regressed["seq"]=3; regressed["revives"]=1;
|
||||
must_fail([&]{queue.store(regressed.dump());});
|
||||
queue.store(snapshot.dump());
|
||||
assert(queue.load().size() == 1 && queue.load()[0].body == snapshot.dump());
|
||||
queue.erase("second-event");
|
||||
queue.prepare(reserve("normal")); queue.prepare(reserve("normal"));
|
||||
assert(queue.load().size() == 1);
|
||||
must_fail([&]{queue.begin("normal");});
|
||||
must_fail([&]{queue.authorize("normal",601);});
|
||||
queue.authorize("normal",60); queue.authorize("normal",60);
|
||||
must_fail([&]{queue.authorize("normal",120);});
|
||||
const auto now = journal::clock::now();
|
||||
queue.begin("normal",now); queue.begin("normal",now+20s);
|
||||
assert(queue.remaining("normal",now+1250ms) == 58750);
|
||||
assert(queue.remaining("normal",now+59999ms) == 1);
|
||||
assert(queue.remaining("normal",now+60s) == 0);
|
||||
assert(queue.remaining("normal",now+61s) == 0);
|
||||
assert(queue.checkpoint("normal",now+1250ms) == 1); // Does not expire a second early.
|
||||
assert(queue.checkpoint("normal",now+59001ms) == 59);
|
||||
must_fail([&]{queue.recover("match");}); // Cannot refund still-consumable authority.
|
||||
queue.stop("normal","manual",now+59500ms); queue.stop("normal","manual",now+120s);
|
||||
assert(queue.load().size() == 1 && queue.load()[0].action == "settle");
|
||||
auto settlement = nlohmann::json::parse(queue.load()[0].body);
|
||||
assert(settlement.at("consumed_seconds") == 60 && settlement.at("end_reason") == "manual");
|
||||
queue.erase(queue.load()[0].id);
|
||||
queue.prepare(reserve("expire")); queue.authorize("expire",60); queue.begin("expire",now);
|
||||
assert(queue.checkpoint("expire",now+60s) == 60);
|
||||
queue.stop("expire","expired",now+61s);
|
||||
assert(nlohmann::json::parse(queue.load()[0].body).at("consumed_seconds") == 60);
|
||||
queue.erase(queue.load()[0].id);
|
||||
queue.prepare(reserve("cancel")); queue.stop("cancel","never_started");
|
||||
assert(nlohmann::json::parse(queue.load()[0].body).at("consumed_seconds") == 0);
|
||||
queue.erase(queue.load()[0].id);
|
||||
// Interrupted temp writes preserve the prior durable snapshot.
|
||||
const auto file = fs::directory_iterator(root)->path();
|
||||
queue.prepare(reserve("failure")); queue.authorize("failure",60); queue.begin("failure",now);
|
||||
fs::create_directory(file.string()+".tmp");
|
||||
must_fail([&]{queue.stop("failure","manual",now+1500ms);});
|
||||
assert(queue.remaining("failure",now+30s) == 0);
|
||||
must_fail([&]{queue.checkpoint("failure",now+30s);});
|
||||
must_fail([&]{queue.begin("failure",now+30s);});
|
||||
fs::remove(file.string()+".tmp");
|
||||
queue.stop("failure","disconnect",now+30s); // A write retry never extends billing.
|
||||
settlement = nlohmann::json::parse(queue.load()[0].body);
|
||||
assert(settlement.at("consumed_seconds") == 2 && settlement.at("end_reason") == "manual");
|
||||
queue.erase(queue.load()[0].id);
|
||||
auto valid = nlohmann::json{{"ok",true},{"state","reserved"},{"request_id","one"},{"code","ok"},
|
||||
{"session_id","session"},{"free_seconds",1200},{"saved_seconds",0},{"available_seconds",1200},
|
||||
{"reserved_seconds",600},{"capacity_seconds",21600},{"authorized_seconds",600},
|
||||
{"bank_balance",10000},{"purchase_points",10000},{"purchase_seconds",600},{"max_purchase_packs",36}};
|
||||
assert(parse_acknowledgement(valid.dump()).valid);
|
||||
for (const auto& field : {"free_seconds","saved_seconds","available_seconds","reserved_seconds","capacity_seconds",
|
||||
"authorized_seconds","bank_balance","purchase_points","purchase_seconds","max_purchase_packs"})
|
||||
for (const auto& invalid : {nlohmann::json(-1),nlohmann::json(1.5),nlohmann::json("10"),nlohmann::json(1000000001)})
|
||||
{ auto value=valid; value[field]=invalid; assert(!parse_acknowledgement(value.dump()).valid); }
|
||||
for (const auto& text : {"", "<html>private-token</html>", "{", "null", "{}"})
|
||||
{ const auto bad=parse_acknowledgement(text); assert(!bad.valid && bad.code.empty() && bad.id.empty()); }
|
||||
assert(!parse_acknowledgement(std::string(0x5000,' ')+valid.dump()).valid);
|
||||
auto invalid = valid; invalid["ok"] = false; assert(!parse_acknowledgement(invalid.dump()).valid);
|
||||
invalid=valid; invalid["session_id"]=""; assert(!parse_acknowledgement(invalid.dump()).valid);
|
||||
invalid=valid; invalid["authorized_seconds"]=604801; assert(!parse_acknowledgement(invalid.dump()).valid);
|
||||
valid["state"]="unresolved"; valid["consumed_seconds"]=nullptr; assert(parse_acknowledgement(valid.dump()).valid);
|
||||
valid["state"]="corrected"; valid["consumed_seconds"]=5; assert(parse_acknowledgement(valid.dump()).valid);
|
||||
{ std::ofstream stream(file); stream << "private-corrupt-journal"; }
|
||||
must_fail([&]{queue.load();}); must_fail([&]{queue.store(request("blocked","event"));});
|
||||
must_fail([&]{queue.recover("new-match");}); must_fail([&]{queue.erase("absent");});
|
||||
fs::remove(file);
|
||||
std::cout << "AFK immutable requests, authorization, elapsed clocks, stop retries and guarded recovery passed\n";
|
||||
}
|
||||
Reference in New Issue
Block a user