10namespace jc = detail::journal_codec;
16 std::array<char, 256> bytes{};
17 std::size_t length = 0;
18 Path() { append(
"event"); }
19 void append(std::string_view value) {
20 if (
value.size() > bytes.size() - length)
22 "admission field path");
23 std::memcpy(bytes.data() + length,
value.data(),
value.size());
24 length +=
value.size();
26 Path child(
const char* name)
const {
28 out.append(
"."); out.append(name);
31 Path element(std::uint64_t index)
const {
34 std::array<char, 20> digits{};
35 std::size_t count = 0;
36 do { digits[count++] = char(
'0' + index % 10); index /= 10; }
while (index);
37 while (count) out.append(std::string_view(&digits[--count], 1));
41 bool is(std::string_view value)
const noexcept {
42 return value == std::string_view(bytes.data(), length);
45template<
class T>
constexpr std::uint64_t tag() {
46 if constexpr (std::is_same_v<T, std::string>)
return 3;
47 else if constexpr (std::is_same_v<T, double>)
return 2;
48 else if constexpr (std::is_same_v<T, bool> || std::is_unsigned_v<T>)
return 0;
53 static constexpr bool loading =
false;
54 static constexpr bool owning =
false;
56 bool counting =
false;
57 std::uint64_t rows = 0;
58 std::uint64_t* next_offset =
nullptr;
59 std::uint64_t* next_row =
nullptr;
60 std::array<std::uint64_t, 4>* array_elements =
nullptr;
61 std::array<std::uint64_t, 4>* array_rows =
nullptr;
62 std::size_t arrays = 0;
64 bool good()
const {
return out.checked_status() == Status::ok && !stopped; }
65 template<
class T>
void field(
const Path& path,
const char* name, T& value) {
67 const auto row = rows++;
68 if (counting || (next_row && row < *next_row))
return;
69 if (next_offset && out.position >= out.offset + out.destination.size) {
73 const auto key = path.child(name);
74 out.word(
key.length); out.bytes(
key.bytes.data(),
key.length);
76 if constexpr (std::is_same_v<T, std::string>) out.text(value);
77 else if constexpr (std::is_same_v<T, double>) {
78 std::uint64_t bits = 0; std::memcpy(&bits, &value,
sizeof bits); out.word(bits);
79 }
else out.word(
static_cast<std::uint64_t
>(value));
80 if (next_offset && out.checked_status() == Status::ok) {
81 if (out.position <= out.offset + out.destination.size) {
82 *next_offset = out.position;
84 }
else stopped =
true;
87 std::uint64_t begin_array(std::size_t slot) {
88 if (!array_elements || !good())
return 0;
91 rows = std::max(rows, (*array_rows)[slot]);
92 return (*array_elements)[slot];
94 void end_element(std::size_t slot, std::uint64_t next) {
95 if (!array_elements || !good())
return;
96 (*array_elements)[slot] = next;
97 (*array_rows)[slot] = rows;
101class FoldSource final :
public detail::ScratchByteSource {
102 const detail::ScratchByteSource& source_;
104 mutable Status failure = Status::ok;
105 mutable std::optional<detail::ReportScratchIoError> carried;
106 explicit FoldSource(
const detail::ScratchByteSource& source) : source_(source) {}
107 std::uint64_t size() const noexcept
override {
return source_.size(); }
108 Status read(std::uint64_t offset, detail::ScratchBuffer out)
const noexcept override {
109 const auto status = source_.read(offset,out);
110 if (status != Status::ok && failure == Status::ok) {
112 if (status == Status::io_error)
113 if (
const auto* error = source_.carried_failure()) carried = *error;
118struct CanonicalFold {
119 std::uint64_t digest;
120 std::uint64_t after_sequence;
122 bool validation_skipped =
false;
123 void bytes(
const char* data, std::size_t size)
noexcept {
124 for (std::size_t i = 0; i != size; ++i)
125 digest = (digest ^
static_cast<unsigned char>(data[i])) * 1099511628211ULL;
127 void word(std::uint64_t value)
noexcept {
128 for (
unsigned i = 0; i != 8; ++i)
129 digest = (digest ^ ((value >> (8 * i)) & 255)) * 1099511628211ULL;
131 template<
class T>
void scalar(
const T& value)
noexcept {
132 if constexpr (std::is_same_v<T, double>) {
133 std::uint64_t bits = 0;
134 std::memcpy(&bits, &value,
sizeof bits);
135 if ((bits & 0x7fffffffffffffffULL) == 0) bits = 0;
136 else if ((bits & 0x7ff0000000000000ULL) == 0x7ff0000000000000ULL
137 && (bits & 0x000fffffffffffffULL)) bits = 0x7ff8000000000000ULL;
139 }
else word(
static_cast<std::uint64_t
>(value));
142constexpr std::size_t kJournalFoldWorkspaceLimit = 1024;
143constexpr std::size_t kJournalFoldFieldBytes = 256;
149constexpr std::size_t kJournalFoldWorkspaceBytes =
sizeof(CanonicalFold) +
sizeof(FoldSource)
151 + 8 *
sizeof(std::uint64_t) + kJournalFoldFieldBytes;
152static_assert(kJournalFoldWorkspaceBytes <= kJournalFoldWorkspaceLimit,
153 "fixed admission fold workspace");
154static_assert(std::is_nothrow_copy_constructible_v<detail::ReportScratchIoError>
155 && std::is_nothrow_copy_assignable_v<detail::ReportScratchIoError>,
156 "diagnostic transport must be nonthrowing");
158template<
bool Owning,
bool Folding = false>
struct Load {
159 static constexpr bool loading =
true;
160 static constexpr bool owning = Owning;
162 std::uint64_t rows = 0, expected_rows = 0;
163 JournalRecordSummary summary;
166 CanonicalFold* fold =
nullptr;
167 bool good()
const {
return in.status == Status::ok; }
168 template<
class T>
void field(
const Path& path,
const char* name, T& value) {
170 if (++rows > expected_rows) { in.status = Status::corrupt;
return; }
171 const auto key = path.child(name);
172 const auto length = in.word();
174 if (length !=
key.length) { in.status = Status::corrupt;
return; }
175 std::array<char, kJournalFoldFieldBytes> bytes{};
176 in.bytes(bytes.data(),
key.length);
178 if (std::memcmp(bytes.data(),
key.bytes.data(),
key.length) || in.word() != tag<T>()) {
179 in.status = Status::corrupt;
return;
181 if constexpr (Folding) {
183 fold->word(length); fold->bytes(bytes.data(),
key.length); fold->word(tag<T>());
186 if constexpr (std::is_same_v<T, std::string>) {
187 if constexpr (!Folding) in.text(value, Owning);
189 const auto size = in.word();
191 if (size > in.remaining() || size >
value.max_size()) {
192 in.status = Status::corrupt;
return;
194 if (!fold->active) in.skip(size);
197 const auto end = in.position + size;
198 if (end > in.cached_at + in.cached_size) fold->validation_skipped =
true;
199 while (in.position != end && good()) {
200 const auto count =
static_cast<std::size_t
>(
201 std::min<std::uint64_t>(bytes.size(), end - in.position));
202 in.bytes(bytes.data(), count);
203 if (good()) fold->bytes(bytes.data(), count);
208 else if constexpr (std::is_same_v<T, double>) in.real(value);
209 else if constexpr (std::is_same_v<T, bool>) {
210 const auto raw = in.word();
211 if (raw > 1) in.status = Status::corrupt;
213 }
else if constexpr (std::is_enum_v<T>) {
214 std::underlying_type_t<T> raw{}; in.integer(raw);
value =
static_cast<T
>(raw);
215 }
else in.integer(value);
216 if constexpr (std::is_integral_v<T> || std::is_enum_v<T>) {
217 if (
key.is(
"event.observation.command") ||
key.is(
"event.receipt.sequence")) {
218 summary.sequence =
static_cast<std::uint64_t
>(
value);
219 if constexpr (Folding)
220 if (summary.sequence <= fold->after_sequence) fold->active =
false;
223 if (
key.is(
"event.observation.bar")) bar =
static_cast<int>(
value);
225 if constexpr (Folding && !std::is_same_v<T, std::string>)
226 if (good() && fold->active) fold->scalar(value);
230template<
class A>
struct Schema {
232 template<
class T>
void field(
const Path& p,
const char* name, T& value) {
233 a.field(p, name, value);
235 void birth(
const Path& p, OrderBirth& value) {
236 auto cause =
value.cause();
auto bar =
value.bar();
auto timestamp =
value.timestamp();
237 auto domain =
value.cursor().domain();
auto position =
value.cursor().position();
238 auto index =
value.cursor().index();
auto count =
value.cursor().count();
240 auto first =
value.first_fill();
auto last =
value.last_fill();
241 auto ordinal =
value.evaluation_ordinal();
242 field(p,
"cause",cause); field(p,
"bar",bar); field(p,
"timestamp",timestamp);
243 field(p,
"cursor_domain",domain); field(p,
"cursor_position",position);
244 field(p,
"cursor_index",index); field(p,
"cursor_count",count);
245 field(p,
"cursor_price",
price); field(p,
"first_fill",first);
246 field(p,
"last_fill",last); field(p,
"evaluation_ordinal",ordinal);
247 if constexpr (A::loading) {
248 if constexpr (A::owning) {
249 if (!a.good())
return;
253 if constexpr (A::owning) value = OrderBirth{};
262 if constexpr (A::owning) {
270 if constexpr (A::owning)
272 "invalid admission birth");
273 else a.in.status = Status::corrupt;
277 void config(
const Path& p, Configuration& x) {
278#define F(name) field(p,#name,x.name)
279 F(process_on_close);
F(calc_on_fills);
F(magnifier);
F(fill_recalculation);
F(scheduler);
280 F(slippage);
F(pyramiding);
F(default_quantity_type);
F(default_quantity_value);
281 F(long_margin);
F(short_margin);
F(commission_value);
F(commission_type);
F(pointvalue);
282 F(fx);
F(quantity_step);
F(mintick);
F(risk_direction);
F(loss_days_limit);
283 F(drawdown_limit);
F(intraday_loss_limit);
F(position_limit);
F(fill_cap_active);
F(risk_halted);
286 void sizing(
const Path& p, SizingObservation& x) {
287 field(p,
"quantity",x.quantity); field(p,
"equity",x.equity); field(p,
"price",x.price);
288 field(p,
"mark",x.mark); field(p,
"fx",x.fx);
290 void observation(
const Path& p, CommandObservation& x) {
291#define F(name) field(p,#name,x.name)
292 F(command);
F(kind); birth(p.child(
"birth"),x.birth);
F(
id);
F(requested_quantity);
293 F(quantity_type);
F(buy); field(p.child(
"prices"),
"limit",x.prices.limit);
294 field(p.child(
"prices"),
"stop",x.prices.stop);
F(oca_name);
F(oca_type);
295 config(p.child(
"configuration"),x.configuration);
F(bar);
F(placement_side);
296 F(placement_cycle);
F(prior_close_quantity);
F(held_quantity);
F(held_entries);
297 F(realized_equity);
F(placement_equity);
F(signal_close);
F(quantized_fixed_quantity);
298 bool present = x.original_sizing.has_value();
299 field(p,
"original_sizing_present",present);
300 if constexpr (A::loading) {
301 if (present) x.original_sizing.emplace();
else x.original_sizing.reset();
303 if (present) sizing(p.child(
"original_sizing"),*x.original_sizing);
304 F(explicit_equity);
F(explicit_price);
307 void receipt(
const Path& p, ReviewReceipt& x) {
308 field(p,
"sequence",x.sequence); field(p,
"checkpoint",x.checkpoint);
309 field(p,
"bar",x.bar); field(p,
"target_command",x.target_command);
311 void receipt(
const Path& p, SizingRevision& x) {
312 field(p,
"sequence",x.sequence); field(p,
"cause_fill",x.cause_fill);
313 field(p,
"bar",x.bar); field(p,
"target_command",x.target_command);
315 void draft(
const Path& p, Draft& x) {
316 bool present = bool(x.observation());
317 field(p,
"observation_present",present);
319 if constexpr (A::loading) {
320 CommandObservation
value;
321 observation(p.child(
"observation"),value);
322 if constexpr (A::owning)
323 if (a.good()) x.bind(std::make_shared<const CommandObservation>(std::move(value)));
324 }
else observation(p.child(
"observation"),
const_cast<CommandObservation&
>(*x.observation()));
326 present = x.review().has_value(); field(p,
"review_present",present);
329 if constexpr (!A::loading) value = *x.review();
330 receipt(p.child(
"review"),value);
331 if constexpr (A::loading && A::owning)
if (a.good()) x.reviewed(value);
333 present = x.sizing_revision().has_value(); field(p,
"sizing_revision_present",present);
335 SizingRevision
value;
336 if constexpr (!A::loading) value = *x.sizing_revision();
337 receipt(p.child(
"sizing_revision"),value);
338 if constexpr (A::loading && A::owning)
if (a.good()) x.sizing_revised(value);
341 void book(
const Path& p, BookObservation& x) {
342#define F(name) field(p,#name,x.name)
343 F(incarnation);
F(priority);
F(bar);
F(type);
F(placement_side);
F(buy);
344 F(
id);
F(oca_name);
F(oca_type); birth(p.child(
"birth"),x.birth);
345 field(p.child(
"prices"),
"limit",x.prices.limit); field(p.child(
"prices"),
"stop",x.prices.stop);
346 field(p.child(
"prices"),
"trail_points",x.prices.trail_points);
347 field(p.child(
"prices"),
"trail_price",x.prices.trail_price);
348 field(p.child(
"prices"),
"trail_offset",x.prices.trail_offset); draft(p.child(
"draft"),x.draft);
351 template<
class T,
class Fn>
void array(
const Path& p, std::vector<T>& rows, Fn emit) {
352 auto count =
static_cast<std::uint64_t
>(rows.size());
353 field(p,
"size",count);
354 if constexpr (A::loading) {
355 if (count > a.expected_rows - std::min(a.rows,a.expected_rows)
356 || count > rows.max_size()) { a.in.status = Status::corrupt;
return; }
357 if constexpr (A::owning) rows.resize(
static_cast<std::size_t
>(count));
359 std::uint64_t first = 0;
360 std::size_t slot = 0;
361 if constexpr (!A::loading) {
363 first = a.begin_array(slot);
365 for (std::uint64_t i = first; i != count && a.good(); ++i) {
366 if constexpr (A::loading && !A::owning) { T row{}; emit(p.element(i),row); }
367 else emit(p.element(i),rows[
static_cast<std::size_t
>(i)]);
368 if constexpr (!A::loading) a.end_element(slot,i+1);
371 void event(
Event& value) {
373 auto kind =
static_cast<std::uint64_t
>(
value.index()); field(p,
"kind",kind);
374 if constexpr (A::loading) {
376 case 0:
value.template emplace<0>();
break;
377 case 1:
value.template emplace<1>();
break;
378 case 2:
value.template emplace<2>();
break;
379 default: a.in.status = Status::corrupt;
return;
382 std::visit([&](
auto& x) {
383 using T = std::decay_t<
decltype(x)>;
384 if constexpr (std::is_same_v<T,CommandEvent>) {
385 if constexpr (A::loading) {
386 CommandObservation
input;
387 observation(p.child(
"observation"),
input);
388 if constexpr (A::owning)
389 if (a.good()) x.observation =
390 std::make_shared<const CommandObservation>(std::move(
input));
392 if (!x.observation) { a.out.status = Status::programming_error;
return; }
393 observation(p.child(
"observation"),
const_cast<CommandObservation&
>(*x.observation));
395 field(p,
"outcome",x.outcome); field(p,
"admitted_incarnation",x.admitted_incarnation);
396 array(p.child(
"before"),x.before,[&](
const auto& q,
auto& row){book(q,row);});
397 array(p.child(
"removed"),x.removed,[&](
const auto& q,
auto&
id){field(q,
"incarnation",id);});
398 }
else if constexpr (std::is_same_v<T,ReviewEvent>) {
399 receipt(p.child(
"receipt"),x.receipt); config(p.child(
"configuration"),x.configuration);
400 field(p,
"open_price",x.open_price); field(p,
"position_side",x.position_side);
401 field(p,
"position_cycle",x.position_cycle);
402 array(p.child(
"book"),x.book,[&](
const auto& q,
auto& row){book(q,row);});
403 array(p.child(
"reviewed"),x.reviewed,[&](
const auto& q,
auto& row){book(q,row);});
404 array(p.child(
"resolutions"),x.resolutions,[&](
const auto& q,
auto& row){
405 field(q,
"incarnation",row.incarnation); field(q,
"kind",row.kind);
406 field(q,
"peer_incarnation",row.peer_incarnation); field(q,
"own_priority",row.own_priority);
407 field(q,
"peer_priority",row.peer_priority); field(q,
"transaction_quantity",row.transaction_quantity);
409 array(p.child(
"causes"),x.causes,[&](
const auto& q,
auto&
id){field(q,
"sequence",id);});
411 receipt(p.child(
"receipt"),x.receipt); field(p,
"incarnation",x.incarnation);
412 sizing(p.child(
"before"),x.before); sizing(p.child(
"after"),x.after);
413 field(p,
"affordability_equity_before",x.affordability_equity_before);
414 field(p,
"affordability_equity_after",x.affordability_equity_after);
420template<
bool Owning,
bool Folding = false>
Status decode(
const detail::ScratchByteSource& source,
422 const detail::ReportScratchIoError*& failure, CanonicalFold* fold =
nullptr) noexcept {
424 if (!
page.data || !
page.size)
return Status::programming_error;
426 jc::Reader in{source,
page};
427 Load<Owning,Folding> load{in};
429 load.expected_rows = in.word();
430 if (load.expected_rows > in.remaining() / 24)
return Status::corrupt;
431 Schema<Load<Owning,Folding>>{load}.event(event);
432 if (load.rows != load.expected_rows && in.status == Status::ok) in.status = Status::corrupt;
433 const auto status = in.finish();
434 if (status == Status::ok) {
435 summary = load.summary;
437 summary.rejected_bar = load.bar;
440 }
catch (
const std::bad_alloc&) {
442 return Status::programming_error;
444 catch (...) {
return Status::corrupt; }
449 jc::observe_work(jc::WorkOperation::payload_source);
452 Save count{counter,
true};
453 Schema<Save>{count}.event(
const_cast<Event&
>(event));
458 Schema<Save>{save}.event(
const_cast<Event&
>(event));
459 length_ = out.position;
460 status_ = out.checked_status();
461 }
catch (...) { status_ = Status::programming_error; }
465 if (status_ != Status::ok)
return status_;
466 if (offset > length_ || buffer.size > length_ - offset || (buffer.size && !buffer.data))
467 return Status::programming_error;
468 if (!buffer.size)
return Status::ok;
469 jc::observe_work(jc::WorkOperation::payload_read, buffer.size);
473 if (offset < next_offset_) {
474 next_offset_ = next_row_ = 0;
475 array_elements_.fill(0);
478 jc::Writer out{next_offset_,offset,buffer};
479 if (!next_offset_) out.word(rows_);
480 Save save{out,
false,0,&next_offset_,&next_row_,&array_elements_,&array_rows_};
481 Schema<Save>{save}.event(
const_cast<Event&
>(*event_));
482 return out.checked_status();
483 }
catch (...) {
return Status::programming_error; }
492 const auto status = decode<true>(
source,page,result,summary,failure);
494 if (status == Status::ok)
event = std::move(result);
496 }
catch (
const std::bad_alloc&) {
498 return Status::programming_error;
505 return decode<false>(
source,page,scratch,summary,failure);
509 std::uint64_t after_sequence, std::optional<detail::ReportScratchIoError>* failure)
noexcept {
510 if (failure) failure->reset();
513 CanonicalFold fold{digest,after_sequence};
516 const auto status = decode<false,true>(
input,page,scratch,candidate,unused,&fold);
517 if (status != Status::ok && fold.validation_skipped) {
525 if (validation != Status::ok) {
526 if (failure && validation == Status::io_error)
527 if (
const auto* error =
source.carried_failure()) *failure = *error;
530 if (failure) *failure =
input.carried;
531 return input.failure != Status::ok ?
input.failure : status;
533 if (failure && status == Status::io_error) *failure =
input.carried;
534 if (status == Status::ok) {
536 if (summary.sequence > after_sequence) digest = fold.digest;
542 if (!page.data || !page.size)
return Status::programming_error;
543 jc::Reader in{
source,page};
544 const auto rows = in.word();
545 if (rows > in.remaining() / 24)
return Status::corrupt;
547 const auto byte = [&](
unsigned value) { h = (h ^ value) * 1099511628211ULL; };
548 const auto word = [&](std::uint64_t value) {
549 for (
unsigned i = 0; i != 8; ++i)
byte((value >> (8 * i)) & 255);
551 const auto text = [&] {
552 const auto length = in.word();
553 if (length > in.remaining()) { in.status = Status::corrupt;
return; }
555 std::array<std::byte, 97> bytes{};
556 for (std::uint64_t at = 0; at < length && in.status == Status::ok;) {
557 const auto count =
static_cast<std::size_t
>(std::min<std::uint64_t>(bytes.size(),length-at));
558 in.bytes(bytes.data(),count);
559 for (std::size_t i = 0; i != count; ++i)
byte(std::to_integer<unsigned>(bytes[i]));
563 for (std::uint64_t i = 0; i != rows && in.status == Status::ok; ++i) {
565 const auto kind = in.word(); word(kind);
566 if (kind == 3) text();
568 auto value = in.word();
570 if ((value & 0x7fffffffffffffffULL) == 0) value = 0;
571 else if ((value & 0x7ff0000000000000ULL) == 0x7ff0000000000000ULL
572 && (value & 0x000fffffffffffffULL)) value = 0x7ff8000000000000ULL;
575 }
else in.status = Status::corrupt;
577 const auto status = in.finish();
578 if (status == Status::ok) digest = h;
static BirthCursor point(BirthCursorDomain domain, int index, int count)
static OrderBirth direct_command(int bar, int64_t timestamp)
static OrderBirth chart_evaluation(int bar, int64_t timestamp)
static OrderBirth fill_evaluation(int bar, int64_t timestamp, BirthCursor cursor, double price, uint64_t first_fill, uint64_t last_fill, uint64_t evaluation_ordinal)
JournalRecordSource(const Event &) noexcept
detail::ScratchStatus read(std::uint64_t, detail::ScratchBuffer) const noexcept override
JournalRecordSource encode_journal_record(const Event &) noexcept
bool rejected_command(Outcome outcome) noexcept
detail::ScratchStatus scan_journal_record(const detail::ScratchByteSource &source, detail::ScratchBuffer page, JournalRecordSummary &summary) noexcept
detail::ScratchStatus decode_journal_record(const detail::ScratchByteSource &, detail::ScratchBuffer, Event &)
detail::ScratchStatus fold_journal_record(const detail::ScratchByteSource &source, detail::ScratchBuffer page, std::uint64_t &digest) noexcept
std::variant< CommandEvent, ReviewEvent, SizingEvent > Event
detail::ScratchStatus fold_validated_journal_record(const detail::ScratchByteSource &source, detail::ScratchBuffer page, JournalRecordSummary &summary, std::uint64_t &digest, std::uint64_t after_sequence, std::optional< detail::ReportScratchIoError > *failure) noexcept
void receipt(const std::string &document, char *json, std::size_t capacity, std::size_t *required)
std::uint64_t page(std::uint64_t bytes)
void rethrow_carried_scratch_failure(const ReportScratchIoError *failure)
const ReportScratchIoError & scratch_out_of_memory_failure() noexcept