Crafter.Review/implementations/Crafter.Review-PgWire.cpp

1218 lines
58 KiB
C++
Raw Normal View History

feat(pipeline): the pure pipeline, checked against its goldens The modules that turn a frozen range into passes and passes into a verdict: :Schema, :Tools (the DENY, git and base lists, a leaf, so the config and the pool both read them), :Plan, :Artifact, :Inventory, :Dispositions, :State, :Normalize, :Stream, :Prompt, :Config, and the URL half of :PgWire. Units, jobs, results, the config and the claude stream stay JSON from end to end, so every key keeps its place; :Text gives JSON values truthiness, get, ==, membership and Str(), and throws Dyn::Error under a name for each kind of unexpected data. Exactness where the bytes are seen: - the stream classifiers are hand-written scanners with case-insensitive folds (the Kelvin sign is a k), Unicode \b and \s, and `.` stopping at \n; the SQL call detection takes `<exe> sql` with any path before it; - config problems carry the JSON scanner's and the UTF-8 decoder's own texts, positions included; - a float total is Neumaier's compensated sum, and the build passes -ffp-contract=off: -march=native would fuse a*b+c; - `crafter-review inventory` and `config` print what the goldens record, byte for byte (the fixture and 14 configs, broken ones included); a message names this binary as the command to run. tests/Goldens: the recorded answers over the fixtures next to them (the commits fixture, a rich versioned series with CRLF, a folded subject, Markdown headings in and out of fences, the code-key rewrites); the suite compares 2304 results as DumpFile writes them, so a key out of order fails too. All agree. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-31 06:48:55 +01:00
// SPDX-License-Identifier: AGPL-3.0-only
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
module;
#include <unistd.h>
#include <cstring>
#include <stdio.h>
feat(pipeline): the pure pipeline, checked against its goldens The modules that turn a frozen range into passes and passes into a verdict: :Schema, :Tools (the DENY, git and base lists, a leaf, so the config and the pool both read them), :Plan, :Artifact, :Inventory, :Dispositions, :State, :Normalize, :Stream, :Prompt, :Config, and the URL half of :PgWire. Units, jobs, results, the config and the claude stream stay JSON from end to end, so every key keeps its place; :Text gives JSON values truthiness, get, ==, membership and Str(), and throws Dyn::Error under a name for each kind of unexpected data. Exactness where the bytes are seen: - the stream classifiers are hand-written scanners with case-insensitive folds (the Kelvin sign is a k), Unicode \b and \s, and `.` stopping at \n; the SQL call detection takes `<exe> sql` with any path before it; - config problems carry the JSON scanner's and the UTF-8 decoder's own texts, positions included; - a float total is Neumaier's compensated sum, and the build passes -ffp-contract=off: -march=native would fuse a*b+c; - `crafter-review inventory` and `config` print what the goldens record, byte for byte (the fixture and 14 configs, broken ones included); a message names this binary as the command to run. tests/Goldens: the recorded answers over the fixtures next to them (the commits fixture, a rich versioned series with CRLF, a folded subject, Markdown headings in and out of fences, the code-key rewrites); the suite compares 2304 results as DumpFile writes them, so a key out of order fails too. All agree. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-31 06:48:55 +01:00
module Crafter.Review:PgWire_impl;
import :PgWire;
import :Json;
import :Text;
import :Crypto;
import :Platform;
import Crafter.Network;
feat(pipeline): the pure pipeline, checked against its goldens The modules that turn a frozen range into passes and passes into a verdict: :Schema, :Tools (the DENY, git and base lists, a leaf, so the config and the pool both read them), :Plan, :Artifact, :Inventory, :Dispositions, :State, :Normalize, :Stream, :Prompt, :Config, and the URL half of :PgWire. Units, jobs, results, the config and the claude stream stay JSON from end to end, so every key keeps its place; :Text gives JSON values truthiness, get, ==, membership and Str(), and throws Dyn::Error under a name for each kind of unexpected data. Exactness where the bytes are seen: - the stream classifiers are hand-written scanners with case-insensitive folds (the Kelvin sign is a k), Unicode \b and \s, and `.` stopping at \n; the SQL call detection takes `<exe> sql` with any path before it; - config problems carry the JSON scanner's and the UTF-8 decoder's own texts, positions included; - a float total is Neumaier's compensated sum, and the build passes -ffp-contract=off: -march=native would fuse a*b+c; - `crafter-review inventory` and `config` print what the goldens record, byte for byte (the fixture and 14 configs, broken ones included); a message names this binary as the command to run. tests/Goldens: the recorded answers over the fixtures next to them (the commits fixture, a rich versioned series with CRLF, a folded subject, Markdown headings in and out of fences, the code-key rewrites); the suite compares 2304 results as DumpFile writes them, so a key out of order fails too. All agree. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-31 06:48:55 +01:00
import std;
namespace Crafter::Review::PgWire {
namespace {
// ---- the statement scanner, on code points ----------------------------------
bool IdentChar(char32_t c) {
return (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '_' || c == '$' || (c >= 0x80 && c <= 0xFFFF);
}
bool TagStart(char32_t c) {
return (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || c == '_' || (c >= 0x80 && c <= 0xFFFF);
}
bool TagChar(char32_t c) {
return TagStart(c) || (c >= '0' && c <= '9');
}
bool At(std::u32string_view s, std::size_t i, std::u32string_view what) {
return s.substr(i).starts_with(what);
}
// DOLLAR_TAG.match(s, i): $tag$ or $$, else empty.
std::u32string_view DollarTag(std::u32string_view s, std::size_t i) {
if (i >= s.size() || s[i] != '$') return {};
std::size_t j = i + 1;
if (j < s.size() && TagStart(s[j])) {
++j;
while (j < s.size() && TagChar(s[j])) ++j;
}
if (j < s.size() && s[j] == '$') return s.substr(i, j + 1 - i);
return {};
}
// The end of a `--` comment: Postgres ends it at \n or \r.
std::size_t EndOfLine(std::u32string_view s, std::size_t i) {
while (i < s.size() && s[i] != '\n' && s[i] != '\r') ++i;
return i;
}
// The end of a `/* */` comment; they nest.
std::size_t EndBlockComment(std::u32string_view s, std::size_t i) {
std::int64_t depth = 0;
while (i < s.size()) {
if (At(s, i, U"/*")) {
++depth;
i += 2;
} else if (At(s, i, U"*/")) {
--depth;
i += 2;
if (depth == 0) return i;
} else {
++i;
}
}
return s.size();
}
std::size_t EndQuote(std::u32string_view s, std::size_t i, char32_t quote, bool backslash) {
++i;
while (i < s.size()) {
if (backslash && s[i] == '\\') {
i += 2;
continue;
}
if (s[i] == quote) {
if (i + 1 < s.size() && s[i + 1] == quote) {
i += 2;
continue;
}
return i + 1;
}
++i;
}
return s.size();
}
// Past whitespace and comments from i.
std::size_t Skip(std::u32string_view s, std::size_t i) {
while (i < s.size()) {
if (Dyn::IsSpace(s[i])) {
++i;
} else if (At(s, i, U"--")) {
i = EndOfLine(s, i);
} else if (At(s, i, U"/*")) {
i = EndBlockComment(s, i);
} else {
break;
}
}
return i;
}
// An E'...' string: a backslash escapes in it.
bool EscapeString(std::u32string_view s, std::size_t i) {
return i > 0 && (s[i - 1] == 'e' || s[i - 1] == 'E') && (i < 2 || !IdentChar(s[i - 2]));
}
constexpr std::array<std::string_view, 5> RoutineWords = {"create", "or", "replace", "function", "procedure"};
// Does the statement start CREATE [OR REPLACE] FUNCTION|PROCEDURE
// (psql's test, on its first four words)?
bool Routine(const std::vector<std::string>& marks) {
auto mark = [&](std::size_t i) { return i < marks.size() ? std::string_view(marks[i]) : std::string_view(); };
auto routine = [](std::string_view w) { return w == "function" || w == "procedure"; };
return mark(0) == "create" && (routine(mark(1)) || (mark(1) == "or" && mark(2) == "replace" && routine(mark(3))));
}
// ---- the wire ---------------------------------------------------------------------
// A failure of the login exchange or of an answer: its kind and its
// text, for "the login exchange broke (KeyError: 'r')".
class Broke : public std::runtime_error {
public:
Broke(std::string type, const std::string& message) : std::runtime_error(message), type(std::move(type)) {}
std::string type;
};
// A socket error, as its message prints: "[Errno 104] Connection reset
// by peer", or "timed out".
class SocketError : public std::runtime_error {
public:
SocketError(const std::string& text, bool timeout) : std::runtime_error(text), timeout(timeout) {}
bool timeout;
};
std::string ConnectErrorText(const std::exception& error) {
if (dynamic_cast<const SocketTimeoutException*>(&error)) return "timed out";
if (const auto* failure = dynamic_cast<const SocketErrorException*>(&error)) {
// "[Errno N] text": the resolver's own text for a resolver
// code, the C library's for an errno.
if (failure->code().category() == ResolverCategory()) return std::format("[Errno {}] {}", failure->code().value(), failure->code().message());
return Dyn::ErrnoText(failure->code().value());
}
return error.what();
}
std::string Be32(std::uint32_t value) {
std::string out(4, '\0');
for (std::int32_t i = 0; i < 4; ++i) out[static_cast<std::size_t>(i)] = static_cast<char>((value >> (24 - 8 * i)) & 0xFF);
return out;
}
std::string Be16(std::uint16_t value) {
return {static_cast<char>(value >> 8), static_cast<char>(value & 0xFF)};
}
std::string CString(std::string_view text) {
return std::format("{}{}", text, std::string(1, '\0'));
}
std::string Message(char type, std::string_view payload = {}) {
return std::format("{}{}{}", type, Be32(static_cast<std::uint32_t>(payload.size() + 4)), payload);
}
std::uint32_t Unpack32(std::string_view bytes, std::size_t at) {
if (at + 4 > bytes.size()) throw Broke("error", "unpack requires a buffer of 4 bytes");
std::uint32_t value = 0;
for (std::size_t i = 0; i < 4; ++i) value = (value << 8) | static_cast<std::uint8_t>(bytes[at + i]);
return value;
}
std::uint16_t Unpack16(std::string_view bytes, std::size_t at) {
if (at + 2 > bytes.size()) throw Broke("error", "unpack requires a buffer of 2 bytes");
return static_cast<std::uint16_t>((static_cast<std::uint8_t>(bytes[at]) << 8) | static_cast<std::uint8_t>(bytes[at + 1]));
}
// bytes.decode(): strict.
std::string Decode(std::string_view bytes) {
if (const std::optional<std::string> error = Dyn::Utf8DecodeError(bytes)) throw Broke("UnicodeDecodeError", *error);
return std::string(bytes);
}
// ErrorResponse / NoticeResponse fields.
Fields FieldsOf(std::string_view payload) {
Fields out;
std::size_t i = 0;
while (i < payload.size() && payload[i] != '\0') {
const std::size_t j = payload.find('\0', i + 1);
if (j == std::string_view::npos) throw Broke("ValueError", "subsection not found");
out[payload[i]] = Dyn::DecodeReplace(payload.substr(i + 1, j - i - 1));
i = j + 1;
}
return out;
}
std::string Get(const Fields& fields, char key, std::string_view fallback = {}) {
const auto found = fields.find(key);
return found == fields.end() ? std::string(fallback) : found->second;
}
// `f.get(a) or f.get(b) or fallback`
std::string FirstOf(const Fields& fields, char a, char b, std::string_view fallback) {
if (const std::string first = Get(fields, a); !first.empty()) return first;
if (const std::string second = Get(fields, b); !second.empty()) return second;
return std::string(fallback);
}
// A SCRAM message's attributes: comma-separated, each cut at its first
// '='.
std::map<std::string, std::string> Attributes(std::string_view text) {
std::map<std::string, std::string> out;
std::size_t n = 0;
for (const std::string& item : Dyn::Split(text, ",")) {
const std::size_t equals = item.find('=');
if (equals == std::string::npos) throw Broke("ValueError", std::format("dictionary update sequence element #{} has length 1; 2 is required", n));
out[item.substr(0, equals)] = item.substr(equals + 1);
++n;
}
return out;
}
const std::string& Attribute(const std::map<std::string, std::string>& attributes, const std::string& key) {
const auto found = attributes.find(key);
if (found == attributes.end()) throw Broke("KeyError", Dyn::Repr(key));
return found->second;
}
std::string Base64(std::string_view text) {
try {
return Base64Decode(text);
} catch (const std::invalid_argument& error) {
throw Broke("Error", error.what());
}
}
constexpr std::array<std::string_view, 2> RetrySqlStates = {"53300", "57P03"};
}
// ---- the URL ---------------------------------------------------------------------------
feat(pipeline): the pure pipeline, checked against its goldens The modules that turn a frozen range into passes and passes into a verdict: :Schema, :Tools (the DENY, git and base lists, a leaf, so the config and the pool both read them), :Plan, :Artifact, :Inventory, :Dispositions, :State, :Normalize, :Stream, :Prompt, :Config, and the URL half of :PgWire. Units, jobs, results, the config and the claude stream stay JSON from end to end, so every key keeps its place; :Text gives JSON values truthiness, get, ==, membership and Str(), and throws Dyn::Error under a name for each kind of unexpected data. Exactness where the bytes are seen: - the stream classifiers are hand-written scanners with case-insensitive folds (the Kelvin sign is a k), Unicode \b and \s, and `.` stopping at \n; the SQL call detection takes `<exe> sql` with any path before it; - config problems carry the JSON scanner's and the UTF-8 decoder's own texts, positions included; - a float total is Neumaier's compensated sum, and the build passes -ffp-contract=off: -march=native would fuse a*b+c; - `crafter-review inventory` and `config` print what the goldens record, byte for byte (the fixture and 14 configs, broken ones included); a message names this binary as the command to run. tests/Goldens: the recorded answers over the fixtures next to them (the commits fixture, a rich versioned series with CRLF, a folded subject, Markdown headings in and out of fences, the code-key rewrites); the suite compares 2304 results as DumpFile writes them, so a key out of order fails too. All agree. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-31 06:48:55 +01:00
Url ParseUrl(std::string_view text) {
const Dyn::SplitUrl split = Dyn::UrlSplit(text);
Url url;
url.scheme = split.scheme;
url.host = Dyn::Lower(split.Hostname().value_or(""));
const std::optional<std::int32_t> port = split.Port();
url.port = port && *port != 0 ? *port : 5432;
const std::optional<std::string> user = split.Username();
url.user = Dyn::Unquote(user && !user->empty() ? *user : "postgres");
url.password = Dyn::Unquote(split.Password().value_or(""));
const std::string path = split.path.empty() ? "/" : split.path;
const std::string database = Dyn::LStrip(path, "/");
url.database = Dyn::Unquote(database.empty() ? "postgres" : database);
return url;
}
std::optional<std::string> CheckUrl(const Json& url) {
if (!url.is_string() || url.get_ref<const std::string&>().empty()) return "no database URL";
Url parsed;
try {
parsed = ParseUrl(url.get_ref<const std::string&>());
} catch (const std::invalid_argument& error) {
return std::format("not a URL ({})", error.what());
}
if (parsed.scheme != "postgres" && parsed.scheme != "postgresql") return "not a postgres:// or postgresql:// URL";
if (!std::ranges::contains(Loopback, parsed.host)) {
const std::vector<std::string> names(Loopback.begin(), Loopback.end());
return std::format("host {} is not this machine ({}): the database under test is a local stack, never a hosted one",
parsed.host.empty() ? "(none)" : parsed.host, Dyn::Join(names, ", "));
}
return std::nullopt;
}
std::string Redact(std::string_view url) {
try {
const Dyn::SplitUrl split = Dyn::UrlSplit(url);
if (!split.Password()) return std::string(url);
const std::string host = split.netloc.substr(split.netloc.rfind('@') + 1);
const std::string netloc = std::format("{}:***@{}", split.Username().value_or(""), host);
return Dyn::UrlUnsplit(split.scheme, netloc, split.path, split.query, split.fragment);
} catch (const std::invalid_argument&) {
return "(unparsable URL)";
}
}
std::string Where(std::string_view url) {
const Url parsed = ParseUrl(url);
return std::format("{} on {}:{} as {}", parsed.database, parsed.host, parsed.port, parsed.user);
}
// ---- statements ----------------------------------------------------------------------
std::vector<std::string> SplitStatements(std::string_view text) {
const std::u32string sql = Dyn::Decode(text);
const std::u32string_view s(sql);
const std::size_t n = s.size();
std::vector<std::u32string_view> pieces;
std::size_t start = 0;
std::size_t i = 0;
std::int64_t depth = 0;
// psql's identifiers[], identifier_count and begin_depth, per
// statement.
std::vector<std::string> marks;
std::int64_t words = 0;
std::int64_t block = 0;
while (i < n) {
const char32_t c = s[i];
const bool letter = (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || c == '_' || c > 127;
if (letter && !(i > 0 && IdentChar(s[i - 1])) && !At(s, i, U"e'") && !At(s, i, U"E'")) {
std::size_t j = i + 1;
while (j < n && IdentChar(s[j])) ++j;
const std::string word = Dyn::Lower(Dyn::Encode(s.substr(i, j - i)));
if (words < 4) marks.push_back(std::ranges::contains(RoutineWords, word) ? word : std::string());
++words;
if (depth == 0 && Routine(marks)) {
if (word == "begin") {
++block;
} else if (word == "case" && block >= 1) {
++block;
} else if (word == "end" && block > 0) {
--block;
}
}
i = j;
continue;
}
if (At(s, i, U"--")) {
i = EndOfLine(s, i);
continue;
}
if (At(s, i, U"/*")) {
i = EndBlockComment(s, i);
continue;
}
if (c == '\'') {
i = EndQuote(s, i, '\'', EscapeString(s, i));
continue;
}
if (c == '"') {
i = EndQuote(s, i, '"', false);
continue;
}
if (c == '$' && !(i > 0 && IdentChar(s[i - 1]))) {
const std::u32string_view tag = DollarTag(s, i);
if (!tag.empty()) {
const std::size_t close = s.find(tag, i + tag.size());
i = close == std::u32string_view::npos ? n : close + tag.size();
continue;
}
}
if (c == '(') {
++depth;
} else if (c == ')') {
depth = std::max<std::int64_t>(0, depth - 1);
} else if (c == ';' && depth == 0 && block == 0) {
pieces.push_back(s.substr(start, i - start));
start = i + 1;
marks.clear();
words = 0;
block = 0;
}
++i;
}
pieces.push_back(s.substr(start));
std::vector<std::string> out;
for (const std::u32string_view piece : pieces) {
if (Skip(piece, 0) < piece.size()) out.push_back(Dyn::Strip(Dyn::Encode(piece)));
}
return out;
}
std::string StripComments(std::string_view text) {
const std::u32string sql = Dyn::Decode(text);
const std::u32string_view s(sql);
std::u32string out;
std::size_t i = 0;
while (i < s.size()) {
if (At(s, i, U"--")) {
i = EndOfLine(s, i);
continue;
}
if (At(s, i, U"/*")) {
i = EndBlockComment(s, i);
continue;
}
const char32_t c = s[i];
std::optional<std::size_t> end;
if (c == '\'') {
end = EndQuote(s, i, '\'', EscapeString(s, i));
} else if (c == '"') {
end = EndQuote(s, i, '"', false);
} else if (c == '$' && !(i > 0 && IdentChar(s[i - 1]))) {
const std::u32string_view tag = DollarTag(s, i);
if (!tag.empty()) {
const std::size_t close = s.find(tag, i + tag.size());
end = close == std::u32string_view::npos ? s.size() : close + tag.size();
}
}
if (!end) {
out.push_back(c);
++i;
} else {
out.append(s.substr(i, *end - i));
i = *end;
}
}
return Dyn::Encode(out);
}
std::string Fingerprint(std::string_view sql) {
std::u32string kept;
for (const char32_t c : Dyn::Decode(StripComments(sql))) {
if (!Dyn::IsSpace(c) && c != ';') kept.push_back(c);
}
return Sha256Hex(Dyn::Encode(kept));
}
std::vector<std::string> FirstWords(std::string_view text, std::size_t k) {
const std::u32string statement = Dyn::Decode(text);
const std::u32string_view s(statement);
std::vector<std::string> words;
std::size_t i = 0;
auto wordStart = [](char32_t c) { return (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || c == '_'; };
auto wordChar = [&](char32_t c) { return wordStart(c) || (c >= '0' && c <= '9') || c == '$'; };
while (words.size() < k) {
i = Skip(s, i);
while (i < s.size() && s[i] == '(' && words.empty()) i = Skip(s, i + 1);
if (i >= s.size() || !wordStart(s[i])) break;
std::size_t j = i + 1;
while (j < s.size() && wordChar(s[j])) ++j;
std::string word = Dyn::Encode(s.substr(i, j - i));
for (char& ch : word) {
if (ch >= 'A' && ch <= 'Z') ch = static_cast<char>(ch - 'A' + 'a');
}
words.push_back(std::move(word));
i = j;
}
return words;
}
std::optional<std::string> Refused(std::string_view statement) {
const std::vector<std::string> w = FirstWords(statement, 4);
const std::string k = w.empty() ? std::string() : w[0];
std::span<const std::string> rest = w.empty() ? std::span<const std::string>() : std::span<const std::string>(w).subspan(1);
if (k == "rollback" && !rest.empty() && (rest[0] == "work" || rest[0] == "transaction")) rest = rest.subspan(1);
const bool to = !rest.empty() && rest[0] == "to";
if (k == "begin" || k == "start" || k == "commit" || k == "end" || k == "abort" || (k == "rollback" && !to)
|| (k == "prepare" && !rest.empty() && rest[0] == "transaction")) {
return std::format("{} - the helper owns the transaction and rolls it back at the end of the call (SAVEPOINT and ROLLBACK TO are fine)", Dyn::Upper(k));
}
if (k == "copy") return "COPY - it reads or writes files, programs or the client outside the transaction; use SELECT / INSERT";
return std::nullopt;
}
std::string Headline(std::string_view statement, std::size_t width) {
const std::string line = Dyn::Join(Dyn::Split(statement), " ");
if (Dyn::Length(line) <= width) return line;
return std::format("{}…", Dyn::Slice(line, 0, static_cast<std::int64_t>(width) - 1));
}
// ---- the connection ----------------------------------------------------------------------
SqlError::SqlError(Fields f) : std::runtime_error(Get(f, 'M', "error")), fields(std::move(f)) {}
struct Connection::State {
Url url;
std::unique_ptr<ClientTCP> socket;
std::string buffer;
std::size_t position = 0;
char status = '\0';
std::vector<Fields> notices;
std::map<std::string, std::string> parameters;
// `count` bytes, or fewer when the server closed first.
std::string ReadExact(std::size_t count) {
while (buffer.size() - position < count) {
if (position > 0 && position == buffer.size()) {
buffer.clear();
position = 0;
}
std::array<char, 65536> chunk {};
try {
const std::uint32_t got = socket->RecieveSync(static_cast<std::uint32_t>(chunk.size()), chunk.data());
buffer.append(chunk.data(), got);
} catch (const SocketClosedException&) {
break;
} catch (const SocketTimeoutException&) {
throw SocketError("timed out", true);
} catch (const SocketErrorException& error) {
throw SocketError(Dyn::ErrnoText(error.code().value()), false);
}
}
const std::size_t take = std::min(count, buffer.size() - position);
std::string out = buffer.substr(position, take);
position += take;
return out;
}
std::pair<char, std::string> Read() {
const std::string header = ReadExact(5);
if (header.size() < 5) throw DbError("the server closed the connection");
const std::uint32_t length = Unpack32(header, 1);
const std::size_t bodyLength = length >= 4 ? length - 4 : 0;
std::string body = ReadExact(bodyLength);
if (body.size() < bodyLength) throw DbError("the server closed the connection");
return {header[0], std::move(body)};
}
void Send(std::string_view data) {
try {
socket->Send(data.data(), static_cast<std::uint32_t>(data.size()));
} catch (const SocketErrorException& error) {
throw DbError(std::format("the connection broke ({})", Dyn::ErrnoText(error.code().value())));
} catch (const SocketClosedException&) {
throw DbError(std::format("the connection broke ({})", Dyn::ErrnoText(32)));
}
}
void Authenticate() {
std::optional<Scram> scram;
while (true) {
std::pair<char, std::string> message;
try {
message = Read();
} catch (const SocketError& error) {
throw DbError(std::format("no answer at login ({})", error.what()));
}
const auto& [type, body] = message;
if (type == 'E') {
const Fields fields = FieldsOf(body);
throw DbError(Dyn::Strip(std::format("login refused: {} {}", Get(fields, 'C'), Get(fields, 'M'))));
}
if (type == 'R') {
const std::uint32_t code = Unpack32(body, 0);
if (code == 0) continue;
if (code == 3) {
Send(Message('p', CString(url.password)));
} else if (code == 5) {
const std::string inner = Md5Hex(url.password + url.user);
const std::string salt = body.substr(4, 4);
Send(Message('p', CString(std::format("md5{}", Md5Hex(inner + salt)))));
} else if (code == 10) {
Json mechanisms = Json::array();
bool scramOffered = false;
for (const std::string& mechanism : Dyn::Split(std::string_view(body).substr(4), std::string_view("\0", 1))) {
if (mechanism.empty()) continue;
mechanisms.push_back(Decode(mechanism));
scramOffered = scramOffered || mechanism == "SCRAM-SHA-256";
}
if (!scramOffered) throw DbError(std::format("unsupported SASL mechanisms {}", Dyn::ReprValue(mechanisms)));
scram.emplace(url.password);
const std::string first = scram->ClientFirst();
Send(Message('p', CString("SCRAM-SHA-256") + Be32(static_cast<std::uint32_t>(first.size())) + first));
} else if (code == 11) {
if (!scram) throw Broke("AttributeError", "'NoneType' object has no attribute 'client_final'");
Send(Message('p', scram->ClientFinal(Decode(std::string_view(body).substr(4)))));
} else if (code == 12) {
if (!scram) throw Broke("AttributeError", "'NoneType' object has no attribute 'verify'");
if (!scram->Verify(Decode(std::string_view(body).substr(4)))) throw DbError("the server failed SCRAM verification");
} else {
throw DbError(std::format("unsupported authentication method {}", code));
}
} else if (type == 'S') {
const std::vector<std::string> parts = Dyn::Split(body, std::string_view("\0", 1));
if (parts.size() < 2) throw Broke("ValueError", std::format("not enough values to unpack (expected 2, got {})", parts.size()));
parameters[Decode(parts[0])] = Decode(parts[1]);
} else if (type == 'Z') {
if (body.empty()) throw Broke("IndexError", "index out of range");
status = body[0];
return;
} else if (type == 'N') {
notices.push_back(FieldsOf(body));
}
}
}
Answer Collect() {
Answer answer;
while (true) {
std::pair<char, std::string> message;
try {
message = Read();
} catch (const SocketError& error) {
if (error.timeout) throw DbError("no answer within the time budget");
throw DbError(std::format("the connection broke ({})", error.what()));
}
const auto& [type, body] = message;
if (type == 'T') {
const std::uint16_t count = Unpack16(body, 0);
std::size_t i = 2;
std::vector<std::string> columns;
for (std::uint16_t c = 0; c < count; ++c) {
const std::size_t end = body.find('\0', i);
if (end == std::string::npos) throw Broke("ValueError", "subsection not found");
columns.push_back(Dyn::DecodeReplace(std::string_view(body).substr(i, end - i)));
i = end + 1 + 18;
}
answer.columns = std::move(columns);
} else if (type == 'D') {
const std::uint16_t count = Unpack16(body, 0);
std::size_t i = 2;
std::vector<std::optional<std::string>> row;
for (std::uint16_t c = 0; c < count; ++c) {
const auto length = static_cast<std::int32_t>(Unpack32(body, i));
i += 4;
if (length < 0) {
row.emplace_back(std::nullopt);
} else {
row.emplace_back(Dyn::DecodeReplace(std::string_view(body).substr(std::min(i, body.size()), static_cast<std::size_t>(length))));
i += static_cast<std::size_t>(length);
}
}
answer.rows.push_back(std::move(row));
} else if (type == 'C') {
answer.tag = Decode(body.empty() ? std::string_view() : std::string_view(body).substr(0, body.size() - 1));
} else if (type == 'I') {
if (!answer.tag || answer.tag->empty()) answer.tag = "(empty)";
} else if (type == 'E') {
answer.error = FieldsOf(body);
} else if (type == 'N') {
answer.notices.push_back(FieldsOf(body));
} else if (type == 'G') {
// COPY FROM STDIN: refused before sending, kept for a server
// that starts one anyway.
Send(Message('f', CString("the review-gates helper sends no COPY data")) + Message('S'));
} else if (type == 'Z') {
if (body.empty()) throw Broke("IndexError", "index out of range");
status = body[0];
return answer;
}
// '1' ParseComplete, '2' BindComplete, 'n' NoData, 's'
// PortalSuspended, 'S' ParameterStatus, 'A', 'K', 'H'/'d'/'c':
// nothing to keep.
}
}
Answer TakeAnswer() {
try {
return Collect();
} catch (const Broke& error) {
throw DbError(std::format("a malformed answer ({}: {})", error.type, error.what()));
}
}
};
Connection::Connection(std::string_view url, std::chrono::seconds timeout, std::string_view application) : state(std::make_unique<State>()) {
state->url = ParseUrl(url);
const Url& p = state->url;
try {
state->socket = std::make_unique<ClientTCP>(p.host, static_cast<std::uint16_t>(p.port), std::chrono::seconds(5));
} catch (const std::exception& error) {
throw DbError(std::format("cannot connect to {}:{} ({})", p.host, p.port, ConnectErrorText(error)));
}
// Every failure from here on is a DbError, whatever broke: its callers
// catch DbError alone.
try {
state->socket->SetReceiveTimeout(timeout);
std::string body = Be32(196608);
for (const auto& [key, value] : std::array<std::pair<std::string_view, std::string_view>, 4>{{
{"user", p.user}, {"database", p.database}, {"application_name", application}, {"client_encoding", "UTF8"}}}) {
body += CString(key) + CString(value);
}
body.push_back('\0');
state->Send(Be32(static_cast<std::uint32_t>(body.size() + 4)) + body);
state->Authenticate();
} catch (const DbError&) {
Close();
throw;
} catch (const Broke& error) {
Close();
throw DbError(std::format("the login exchange broke ({}: {})", error.type, error.what()));
}
}
Connection::~Connection() {
Close();
}
Answer Connection::Simple(std::string_view query) {
state->Send(Message('Q', CString(query)));
return state->TakeAnswer();
}
Answer Connection::Extended(std::string_view statement) {
state->Send(Message('P', std::format("{}{}{}", std::string(1, '\0'), CString(statement), Be16(0)))
+ Message('B', std::format("{}{}{}{}", std::string(2, '\0'), Be16(0), Be16(0), Be16(0)))
+ Message('D', std::string("P\0", 2))
+ Message('E', std::format("{}{}", std::string(1, '\0'), Be32(0)))
+ Message('S'));
return state->TakeAnswer();
}
void Connection::Close() {
if (!state || !state->socket) return;
try {
const std::string terminate = Message('X');
state->socket->Send(terminate.data(), static_cast<std::uint32_t>(terminate.size()));
} catch (const std::exception&) {
// Closing anyway.
}
state->socket.reset();
}
char Connection::Status() const {
return state->status;
}
const std::map<std::string, std::string>& Connection::Parameters() const {
return state->parameters;
}
Scram::Scram(std::string password, std::string user, std::string nonce) : password(std::move(password)), nonce(nonce.empty() ? Base64Encode(RandomBytes(18)) : std::move(nonce)) {
bare = std::format("n={},r={}", user, this->nonce);
}
std::string Scram::ClientFirst() const {
return std::format("n,,{}", bare);
}
std::string Scram::ClientFinal(std::string_view serverFirst) {
const std::map<std::string, std::string> a = Attributes(serverFirst);
const std::string& serverNonce = Attribute(a, "r");
if (!serverNonce.starts_with(nonce)) throw DbError("SCRAM: the server nonce does not extend ours");
const std::string salt = Base64(Attribute(a, "s"));
const std::string& iterationText = Attribute(a, "i");
const std::optional<std::int64_t> iterations = Dyn::ParseInt(iterationText);
if (!iterations) throw Broke("ValueError", Dyn::IntError(iterationText));
if (*iterations < 1) throw Broke("ValueError", "iteration value must be greater than 0.");
const std::string salted = Pbkdf2Sha256(password, salt, static_cast<std::uint32_t>(*iterations));
const std::string clientKey = HmacSha256(salted, "Client Key");
const std::string withoutProof = std::format("c={},r={}", Base64Encode("n,,"), serverNonce);
authMessage = std::format("{},{},{}", bare, serverFirst, withoutProof);
const std::string signature = HmacSha256(Sha256(clientKey), authMessage);
serverKey = HmacSha256(salted, "Server Key");
std::string proof(clientKey.size(), '\0');
for (std::size_t i = 0; i < clientKey.size(); ++i) proof[i] = static_cast<char>(clientKey[i] ^ signature[i]);
return std::format("{},p={}", withoutProof, Base64Encode(proof));
}
bool Scram::Verify(std::string_view serverFinal) const {
const std::map<std::string, std::string> a = Attributes(serverFinal);
const auto v = a.find("v");
return v != a.end() && ConstantTimeEqual(Base64(v->second), HmacSha256(serverKey, authMessage));
}
std::unique_ptr<Connection> Connect(std::string_view url, std::int32_t attempts, const std::function<void(double)>& sleep) {
double delay = 0.5;
std::optional<DbError> last;
for (std::int32_t n = 0; n < attempts; ++n) {
try {
return std::make_unique<Connection>(url);
} catch (const DbError& error) {
last = error;
const std::string_view text = error.what();
const bool busy = std::ranges::any_of(RetrySqlStates, [&](std::string_view state) { return text.contains(state); });
if (!busy && (n >= 2 || text.contains("login refused"))) break;
// No sleep after the last attempt: nothing comes after it.
if (n + 1 < attempts) {
if (sleep) {
sleep(delay);
} else {
std::this_thread::sleep_for(std::chrono::duration<double>(delay));
}
}
delay = std::min(4.0, delay * 2);
}
}
if (last) throw *last;
throw DbError("no attempt made");
}
// ---- output ----------------------------------------------------------------------------
std::string Cell(const std::optional<std::string>& value) {
if (!value) return "(null)";
const std::string v = Dyn::Replace(Dyn::Replace(Dyn::Replace(*value, "\r", "\\r"), "\n", "\\n"), "\t", "\\t");
if (Dyn::Length(v) <= static_cast<std::size_t>(MaxCell)) return v;
return std::format("{}…", Dyn::Slice(v, 0, MaxCell - 1));
}
std::string Table(const std::vector<std::string>& columns, const std::vector<std::vector<std::optional<std::string>>>& rows) {
std::vector<std::vector<std::string>> shown;
for (std::size_t r = 0; r < rows.size() && r < static_cast<std::size_t>(MaxRows); ++r) {
std::vector<std::string> cells;
for (const std::optional<std::string>& value : rows[r]) cells.push_back(Cell(value));
shown.push_back(std::move(cells));
}
std::vector<std::size_t> widths;
for (std::size_t i = 0; i < columns.size(); ++i) {
std::size_t width = Dyn::Length(columns[i]);
for (const std::vector<std::string>& row : shown) width = std::max(width, Dyn::Length(row.at(i)));
widths.push_back(width);
}
auto line = [&](const std::vector<std::string>& cells) {
std::vector<std::string> padded;
for (std::size_t i = 0; i < cells.size(); ++i) padded.push_back(Dyn::LJust(cells[i], widths.at(i)));
return Dyn::RStrip(std::format(" {}", Dyn::Join(padded, " | ")));
};
std::vector<std::string> out = {line(columns)};
std::vector<std::string> dashes;
for (const std::size_t width : widths) dashes.emplace_back(width, '-');
out.push_back(std::format("-{}-", Dyn::Join(dashes, "-+-")));
for (const std::vector<std::string>& row : shown) out.push_back(line(row));
const std::size_t more = rows.size() - shown.size();
out.push_back(std::format("({} row{}{})", rows.size(), rows.size() != 1 ? "s" : "", more > 0 ? std::format(", {} not shown", more) : std::string()));
return Dyn::Join(out, "\n");
}
std::string ErrorText(const Fields& f, std::string_view statement) {
std::vector<std::string> lines = {Dyn::RStrip(std::format("{} {}: {}", FirstOf(f, 'V', 'S', "ERROR"), Get(f, 'C'), Get(f, 'M')))};
for (const auto& [key, label] : std::array<std::pair<char, std::string_view>, 3>{{{'D', "DETAIL"}, {'H', "HINT"}, {'W', "WHERE"}}}) {
if (const std::string value = Get(f, key); !value.empty()) lines.push_back(std::format("{}: {}", label, value));
}
const std::string position = Get(f, 'P');
if (!position.empty() && !statement.empty() && std::ranges::all_of(position, [](char c) { return c >= '0' && c <= '9'; })) {
const std::int64_t p = Dyn::ParseInt(position).value_or(1) - 1;
const std::string before = Dyn::Slice(statement, std::max<std::int64_t>(0, p - 30), p);
const std::string after = Dyn::Slice(statement, p, p + 30);
lines.push_back(Dyn::Replace(std::format("AT: …{}⟦{}⟧…", before, after), "\n", " "));
}
return Dyn::Join(lines, "\n");
}
std::string NoticeText(const Fields& f) {
return std::format("{}: {}", FirstOf(f, 'V', 'S', "NOTICE"), Get(f, 'M'));
}
// ---- one call of the helper ------------------------------------------------------------
std::int32_t RunSql(std::string_view url, std::string_view sql, const Out& out, const Connector& connect, const Refuser& refuse) {
if (const std::optional<std::string> problem = CheckUrl(Json(url))) {
out(std::format("{} - {}", NoDatabase, *problem));
return ExitNoDb;
}
if (Dyn::Length(sql) > static_cast<std::size_t>(MaxSql)) {
out(std::format("review-gates sql: the script is {} characters; the limit is {} - nothing ran", Dyn::Length(sql), MaxSql));
return ExitUsage;
}
const std::vector<std::string> statements = SplitStatements(sql);
if (statements.empty()) {
out("review-gates sql: no statement given - the SQL goes on stdin, in a quoted heredoc (<<'SQL' ... SQL) - nothing ran");
return ExitUsage;
}
if (statements.size() > static_cast<std::size_t>(MaxStatements)) {
out(std::format("review-gates sql: {} statements; the limit is {} - nothing ran", statements.size(), MaxStatements));
return ExitUsage;
}
std::vector<std::string> bad;
for (std::size_t i = 0; i < statements.size(); ++i) {
if (const std::optional<std::string> why = refuse ? refuse(statements[i]) : Refused(statements[i])) bad.push_back(std::format("statement {}: {}", i + 1, *why));
}
if (!bad.empty()) {
out(std::format("review-gates sql: refused, nothing ran - {}", Dyn::Join(bad, "; ")));
return ExitSql;
}
std::unique_ptr<Connection> conn;
try {
conn = connect ? connect(url) : Connect(url);
} catch (const DbError& error) {
out(std::format("{} - {}", NoDatabase, error.what()));
return ExitNoDb;
}
std::size_t size = 0;
std::int32_t code = ExitOk;
std::optional<std::size_t> ended;
auto say = [&](const std::string& text, bool always = false) {
if (size < static_cast<std::size_t>(MaxOutput) || always) out(text);
size += Dyn::Length(text) + 1;
};
// Whatever happened after the connection, the transaction is rolled
// back and the call says how it ended.
auto finish = [&] {
try {
if (conn->Status() == 'T' || conn->Status() == 'E') conn->Simple("ROLLBACK");
} catch (const std::exception&) {
// Nothing more to do: the connection closes either way.
}
conn->Close();
out(ended ? std::format("-- the transaction ended at statement {}: what ran up to it may have been committed; the rest was not", *ended)
: std::string("-- rolled back: nothing of this call was committed"));
};
try {
for (const std::string& query : {std::string("BEGIN"), std::format("SET LOCAL statement_timeout = '{}s'", StatementTimeoutS),
std::format("SET LOCAL lock_timeout = '{}s'", LockTimeoutS), std::string("SET LOCAL idle_in_transaction_session_timeout = '60s'")}) {
const Answer answer = conn->Simple(query);
if (answer.error) {
out(std::format("{} - the session could not be set up ({})", NoDatabase, ErrorText(*answer.error)));
finish();
return ExitNoDb;
}
}
out(std::format("{} to {} - ONE transaction, rolled back at the end of this call (statement_timeout {}s, lock_timeout {}s; at most {} rows printed per statement)",
Answered, Where(url), StatementTimeoutS, LockTimeoutS, MaxRows));
for (std::size_t i = 1; i <= statements.size(); ++i) {
const std::string& statement = statements[i - 1];
say(std::format("-- [{}/{}] {}", i, statements.size(), Headline(statement)), true);
const Answer answer = conn->Extended(statement);
for (const Fields& notice : answer.notices) say(NoticeText(notice));
if (answer.error) {
say(ErrorText(*answer.error, statement), true);
if (i < statements.size()) say(std::format("-- stopped: statement {} failed, so statements {}-{} did not run", i, i + 1, statements.size()), true);
code = ExitSql;
break;
}
if (answer.columns) say(Table(*answer.columns, answer.rows));
say(answer.tag && !answer.tag->empty() ? *answer.tag : std::string("(no command tag)"), true);
if (conn->Status() != 'T' && conn->Status() != 'E') {
ended = i;
say(std::format("-- STOPPED: the transaction is no longer open after statement {} (server status {}) - nothing more runs. "
"This is a pipeline defect: report it as a pipeline note; what ran up to it may have been committed.", i, conn->Status()), true);
code = ExitSql;
break;
}
}
if (size > static_cast<std::size_t>(MaxOutput)) {
out(std::format("-- output cut at {} characters ({} produced): the statements ran, their rows are not all shown", MaxOutput, size));
}
} catch (const DbError& error) {
out(std::format("-- the database stopped answering mid-call ({}); nothing of this call was committed", error.what()));
finish();
return ExitSql;
} catch (...) {
finish();
throw;
}
finish();
return code;
}
std::pair<std::vector<std::string>, std::vector<std::vector<std::optional<std::string>>>> Query(std::string_view url, std::string_view sql) {
const std::unique_ptr<Connection> conn = Connect(url, 3);
auto rollback = [&] {
try {
conn->Simple("ROLLBACK");
} catch (const std::exception&) {
// Closing anyway.
}
conn->Close();
};
try {
conn->Simple("BEGIN READ ONLY");
Answer answer = conn->Extended(sql);
if (answer.error) throw SqlError(*answer.error);
rollback();
return {answer.columns.value_or(std::vector<std::string>()), std::move(answer.rows)};
} catch (...) {
rollback();
throw;
}
}
std::optional<bool> BypassesRls(std::string_view url) {
try {
const auto [columns, rows] = Query(url, "select coalesce((select rolsuper or rolbypassrls from pg_roles where rolname = current_user), false)");
return rows.at(0).at(0) == std::optional<std::string>("t");
} catch (const DbError&) {
return std::nullopt;
} catch (const SqlError&) {
return std::nullopt;
}
}
std::pair<bool, std::string> Probe(std::string_view url) {
if (const std::optional<std::string> problem = CheckUrl(Json(url))) return {false, *problem};
std::vector<std::vector<std::optional<std::string>>> rows;
try {
rows = Query(url, "select current_user, current_database(), version()").second;
} catch (const DbError& error) {
return {false, std::format("{}: {}", Redact(url), error.what())};
} catch (const SqlError& error) {
return {false, std::format("{}: {}", Redact(url), error.what())};
}
const std::vector<std::optional<std::string>>& row = rows.at(0);
const std::vector<std::string> version = Dyn::Split(row.at(2).value_or(""));
const std::span<const std::string> head(version.data(), std::min<std::size_t>(version.size(), 2));
const Url parsed = ParseUrl(url);
return {true, std::format("{} on {}:{} as {} - {}", row.at(1).value_or("None"), parsed.host, parsed.port, row.at(0).value_or("None"), Dyn::Join(head, " "))};
}
// ---- the broker ------------------------------------------------------------------------------
struct Broker::State {
Runner run;
Prober probe;
std::mutex lock;
std::map<std::string, std::string> tokens;
std::map<std::string, std::string, std::less<>> urls;
std::unique_ptr<ListenerTCP> listener;
std::thread accepting;
std::uint16_t port = 0;
void Serve(ClientTCP& client) {
bool dead = false;
auto out = [&](const std::string& text) {
if (dead) return;
const std::string line = std::format("{}\n", Dyn::DecodeReplace(text));
try {
client.Send(line.data(), static_cast<std::uint32_t>(line.size()));
} catch (const std::exception&) {
// The helper is gone (its Bash call was stopped): the call
// still runs to its ROLLBACK.
dead = true;
}
};
// rfile.readline(limit): up to a newline, at most `limit` bytes.
constexpr std::size_t Limit = 4 * static_cast<std::size_t>(MaxSql) + 4096;
std::string raw;
while (raw.size() < Limit && !raw.ends_with('\n')) {
std::array<char, 65536> chunk {};
try {
const std::uint32_t got = client.RecieveSync(static_cast<std::uint32_t>(std::min(chunk.size(), Limit - raw.size())), chunk.data());
const std::string_view piece(chunk.data(), got);
const std::size_t newline = piece.find('\n');
raw.append(newline == std::string_view::npos ? piece : piece.substr(0, newline + 1));
} catch (const std::exception&) {
break;
}
}
Json request;
if (raw.ends_with('\n')) {
try {
request = ParseLoose(raw);
} catch (const std::exception&) {
request = nullptr;
}
}
std::int32_t code = ExitOk;
const Json& token = request.is_object() ? Dyn::Get(request, "token") : Dyn::EmptyDict();
std::optional<std::string> url;
if (request.is_object() && token.is_string()) {
const std::lock_guard guard(lock);
const auto found = urls.find(token.get_ref<const std::string&>());
if (found != urls.end()) url = found->second;
}
if (!request.is_object()) {
out(std::format("review-gates sql: the call did not arrive whole (more than {} characters, or not the helper) - nothing ran", MaxSql));
code = ExitUsage;
} else if (!url) {
// A token that is no string (a list, say) gets this answer too.
out(std::format("{} - the driver knows no database by this token (a stale or a foreign one)", NoDatabase));
code = ExitNoDb;
} else if (Dyn::Truthy(Dyn::Get(request, "probe"))) {
const auto [ok, detail] = probe ? probe(*url) : Probe(*url);
out(std::format("{} - {}", ok ? Answered : NoDatabase, detail));
code = ok ? ExitOk : ExitNoDb;
} else {
const Json& sql = Dyn::Get(request, "sql");
const std::string text = sql.is_string() ? sql.get<std::string>() : std::string();
code = run ? run(*url, text, out) : RunSql(*url, text, out);
}
out(std::format("{}{}", ExitMark, code));
}
};
Broker::Broker(Runner run, Prober probe) : state(std::make_shared<State>()) {
state->run = std::move(run);
state->probe = std::move(probe);
}
Broker::~Broker() {
Close();
}
void Broker::Start() {
const std::weak_ptr<State> weak = state;
state->listener = std::make_unique<ListenerTCP>("127.0.0.1", 0, [weak](ClientTCP* accepted) {
std::unique_ptr<ClientTCP> client(accepted);
const std::shared_ptr<State> owner = weak.lock();
if (!owner) return;
// A thread per call, not Crafter.Thread's CPU-sized pool: a call
// waits on the database, not on a CPU.
std::thread([owner, client = std::move(client)] {
try {
owner->Serve(*client);
} catch (const std::exception&) {
// A call that broke ends here; the helper reports the
// missing exit code.
}
}).detach();
});
state->port = state->listener->Port();
ListenerTCP* listener = state->listener.get();
state->accepting = std::thread([listener] { listener->ListenSyncSync(); });
}
std::string Broker::Token(const std::string& url) {
const std::lock_guard guard(state->lock);
const auto found = state->tokens.find(url);
if (found != state->tokens.end()) return found->second;
const std::string token = TokenHex(16);
state->tokens[url] = token;
state->urls[token] = url;
return token;
}
std::optional<std::string> Broker::UrlOf(std::string_view token) {
const std::lock_guard guard(state->lock);
const auto found = state->urls.find(token);
if (found == state->urls.end()) return std::nullopt;
return found->second;
}
std::string Broker::Address(const std::string& url) {
if (url.empty() || !state->listener) return "";
return std::format("127.0.0.1:{}#{}", state->port, Token(url));
}
void Broker::Close() {
if (!state || !state->listener) return;
state->listener->Stop();
if (state->accepting.joinable()) state->accepting.join();
state->listener.reset();
}
std::int32_t CallBroker(std::string_view address, std::string_view sql, bool probeOnly, const Out& out) {
if (address.empty() || !address.contains('#')) {
out(std::format("{} - this pass has no database (the driver gives one to a pass with the supabase environment)", NoDatabase));
return ExitNoDb;
}
const std::size_t hash = address.rfind('#');
const std::string_view where = address.substr(0, hash);
const std::string token(address.substr(hash + 1));
const std::size_t colon = where.rfind(':');
const std::string host(colon == std::string_view::npos ? std::string_view() : where.substr(0, colon));
const std::string portText(colon == std::string_view::npos ? where : where.substr(colon + 1));
const std::optional<std::int64_t> port = Dyn::ParseInt(portText);
if (!port) {
out(std::format("{} - the driver's SQL broker does not answer ({})", NoDatabase, Dyn::IntError(portText)));
return ExitNoDb;
}
// socket.create_connection raised OverflowError for it, which
// call_broker did not catch: the helper ended in a traceback.
if (*port < 0 || *port > 65535) throw Dyn::Error("OverflowError", "getsockaddrarg: port must be 0-65535.");
std::unique_ptr<ClientTCP> socket;
try {
socket = std::make_unique<ClientTCP>(host, static_cast<std::uint16_t>(*port), std::chrono::seconds(5));
} catch (const std::exception& error) {
out(std::format("{} - the driver's SQL broker does not answer ({})", NoDatabase, ConnectErrorText(error)));
return ExitNoDb;
}
try {
// Each statement's lines come within its timeout; connecting may
// retry.
socket->SetReceiveTimeout(std::chrono::seconds(StatementTimeoutS + 40));
const Json request = probeOnly ? Json{{"token", token}, {"probe", true}} : Json{{"token", token}, {"sql", sql}};
const std::string line = std::format("{}\n", DumpSpaced(request));
socket->Send(line.data(), static_cast<std::uint32_t>(line.size()));
std::string pending;
bool closed = false;
while (!closed) {
std::array<char, 65536> chunk {};
try {
const std::uint32_t got = socket->RecieveSync(static_cast<std::uint32_t>(chunk.size()), chunk.data());
pending.append(chunk.data(), got);
} catch (const SocketClosedException&) {
closed = true;
}
std::size_t newline = 0;
while ((newline = pending.find('\n')) != std::string::npos || (closed && !pending.empty())) {
const std::size_t end = newline == std::string::npos ? pending.size() : newline + 1;
std::string text = Dyn::RStrip(Dyn::DecodeReplace(std::string_view(pending).substr(0, end)), "\n");
pending.erase(0, end);
if (text.starts_with(ExitMark)) {
const std::string number = text.substr(ExitMark.size());
const std::optional<std::int64_t> code = Dyn::ParseInt(number);
if (!code) throw std::invalid_argument(Dyn::IntError(number));
return static_cast<std::int32_t>(*code);
}
out(text);
}
}
} catch (const SocketTimeoutException&) {
out("-- the driver's SQL broker stopped answering mid-call (timed out); nothing of this call was committed");
return ExitSql;
} catch (const SocketErrorException& error) {
out(std::format("-- the driver's SQL broker stopped answering mid-call ({}); nothing of this call was committed", Dyn::ErrnoText(error.code().value())));
return ExitSql;
} catch (const std::exception& error) {
out(std::format("-- the driver's SQL broker stopped answering mid-call ({}); nothing of this call was committed", error.what()));
return ExitSql;
}
out("-- the driver's SQL broker closed the call without its exit code; nothing of this call was committed");
return ExitSql;
}
std::int32_t SqlMain(std::span<const std::string_view> args) {
constexpr std::string_view Usage = R"(SQL against the database under test, for a lens with the `supabase` environment - ONE transaction per call, rolled
back at its end.
crafter-review sql <<'SQL' the statements on stdin (a quoted heredoc: the shell expands nothing in it)
select ...;
SQL
crafter-review sql 'select 1' or as arguments
crafter-review sql --probe does the database answer?
The SQL goes to the review driver's broker, named with this pass's database in REVIEW_GATES_DB (an address and a
token - the database's URL stays in the driver); the broker runs it on the local stack and sends the lines back.
Exit 0 every statement ran, 1 one failed or was refused, 2 no database, 64 no statement.)";
const std::string address = Platform::GetEnv(BrokerEnv).value_or("");
const Out say = [](const std::string& line) {
std::println("{}", line);
std::fflush(stdout);
};
if (args.size() == 1 && args[0] == "--probe") return CallBroker(address, "", true, say);
if (!args.empty() && (args[0] == "-h" || args[0] == "--help")) {
std::println("{}", Usage);
return 0;
}
std::string sql;
if (!args.empty()) {
std::vector<std::string> words(args.begin(), args.end());
sql = Dyn::Join(words, " ");
} else if (!isatty(0)) {
// All of stdin, as text with universal newlines.
const std::string bytes((std::istreambuf_iterator<char>(std::cin)), std::istreambuf_iterator<char>());
sql = Dyn::UniversalNewlines(Dyn::DecodeReplace(bytes));
}
return CallBroker(address, sql, false, say);
}
feat(pipeline): the pure pipeline, checked against its goldens The modules that turn a frozen range into passes and passes into a verdict: :Schema, :Tools (the DENY, git and base lists, a leaf, so the config and the pool both read them), :Plan, :Artifact, :Inventory, :Dispositions, :State, :Normalize, :Stream, :Prompt, :Config, and the URL half of :PgWire. Units, jobs, results, the config and the claude stream stay JSON from end to end, so every key keeps its place; :Text gives JSON values truthiness, get, ==, membership and Str(), and throws Dyn::Error under a name for each kind of unexpected data. Exactness where the bytes are seen: - the stream classifiers are hand-written scanners with case-insensitive folds (the Kelvin sign is a k), Unicode \b and \s, and `.` stopping at \n; the SQL call detection takes `<exe> sql` with any path before it; - config problems carry the JSON scanner's and the UTF-8 decoder's own texts, positions included; - a float total is Neumaier's compensated sum, and the build passes -ffp-contract=off: -march=native would fuse a*b+c; - `crafter-review inventory` and `config` print what the goldens record, byte for byte (the fixture and 14 configs, broken ones included); a message names this binary as the command to run. tests/Goldens: the recorded answers over the fixtures next to them (the commits fixture, a rich versioned series with CRLF, a folded subject, Markdown headings in and out of fences, the code-key rewrites); the suite compares 2304 results as DumpFile writes them, so a key out of order fails too. All agree. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-31 06:48:55 +01:00
}