// Matrix Construct // // Copyright (C) Matrix Construct Developers, Authors & Contributors // Copyright (C) 2016-2018 Jason Volk // // Permission to use, copy, modify, and/or distribute this software for any // purpose with or without fee is hereby granted, provided that the above // copyright notice and this permission notice is present in all copies. The // full license for this software is available in the LICENSE file. #include namespace ircd::m::vm { uint64_t last_sequence(id::event::buf &); uint64_t last_sequence(); } decltype(ircd::m::vm::log) ircd::m::vm::log { "vm", 'v' }; decltype(ircd::m::vm::accept) ircd::m::vm::accept {}; decltype(ircd::m::vm::current_sequence) ircd::m::vm::current_sequence {}; decltype(ircd::m::vm::default_opts) ircd::m::vm::default_opts {}; decltype(ircd::m::vm::default_commit_opts) ircd::m::vm::default_commit_opts {}; // // init // ircd::m::vm::init::init() { id::event::buf event_id; current_sequence = last_sequence(event_id); log.info("Initializing from vm sequence %lu [%s]", current_sequence, current_sequence? string_view{event_id} : "NO EVENTS"_sv); } ircd::m::vm::init::~init() { id::event::buf event_id; const auto current_sequence { last_sequence(event_id) }; log.info("Shutting down @ %lu [%s]", current_sequence, current_sequence? string_view{event_id} : "NO EVENTS"_sv); } // // commit // /// This function takes an event object vector and adds our origin and event_id /// and hashes and signature and attempts to inject the event into the core. /// ircd::m::event::id::buf ircd::m::vm::commit(json::iov &event, const json::iov &contents, const opts::commit &opts) { const json::iov::add_if _origin { event, opts.origin, { "origin", my_host() } }; const json::iov::add_if _origin_server_ts { event, opts.origin_server_ts, { "origin_server_ts", ircd::time() } }; const json::strung content { contents }; // event_id sha256::buf event_id_hash; if(opts.event_id) { const json::iov::push _content { event, { "content", content }, }; thread_local char preimage_buf[64_KiB]; event_id_hash = sha256 { stringify(mutable_buffer{preimage_buf}, event) }; } event::id::buf eid_buf; const string_view event_id { opts.event_id? m::event_id(event, eid_buf, event_id_hash): string_view{} }; const json::iov::add_if _event_id { event, opts.event_id, { "event_id", event_id } }; // hashes char hashes_buf[128]; const string_view hashes { opts.hash? m::event::hashes(hashes_buf, event, content): string_view{} }; const json::iov::add_if _hashes { event, opts.hash, { "hashes", hashes } }; // sigs char sigs_buf[384]; const string_view sigs { opts.sign? m::event::signatures(sigs_buf, event, contents): string_view{} }; const json::iov::add_if _sigs { event, opts.sign, { "signatures", sigs } }; const json::iov::push _content { event, { "content", content }, }; commit(event, opts); return eid_buf; } namespace ircd::m::vm { extern hook::site commit_hook; } decltype(ircd::m::vm::commit_hook) ircd::m::vm::commit_hook { { "name", "vm.commit" } }; /// Insert a new event originating from this server. /// /// Figure 1: /// in . <-- injection /// ===:::::::==// /// | ||||||| // <-- this function /// | \\|// //| /// | ||| // | | acceleration /// | |||// | | /// | |||/ | | /// | ||| | V /// | !!! | /// | * | <----- nozzle /// | ///|||\\\ | /// |/|/|/|\|\|\| <---- propagation cone /// _/|/|/|/|\|\|\|\_ /// out /// ircd::m::vm::fault ircd::m::vm::commit(const event &event, const opts::commit &opts) { if(opts.debuglog_precommit) log.debug("injecting event(mark +%ld) %s", vm::current_sequence, pretty_oneline(event)); check_size(event); commit_hook(event); auto opts_{opts}; opts_.verify = false; opts_.reserve_bytes = serialized(event); // Some functionality on this server may create an event on behalf // of remote users. It's safe for us to mask this here, but eval'ing // this event in any replay later will require special casing. opts_.non_conform |= event::conforms::MISMATCH_ORIGIN_SENDER; //TODO: X opts_.non_conform |= event::conforms::MISSING_PREV_STATE; vm::eval eval{opts_}; const fault ret { eval(event) }; if(opts_.infolog_postcommit) log.info("%s @%lu %s", reflect(ret), sequence(eval), pretty_oneline(event, false)); return ret; } // // Eval // // Processes any event from any place from any time and does whatever is // necessary to validate, reject, learn from new information, ignore old // information and advance the state of IRCd as best as possible. namespace ircd::m::vm { extern hook::site eval_hook; extern hook::site notify_hook; void _tmp_effects(const m::event &event); //TODO: X void write(eval &); fault _eval_edu(eval &, const event &); fault _eval_pdu(eval &, const event &); } decltype(ircd::m::vm::eval_hook) ircd::m::vm::eval_hook { { "name", "vm.eval" } }; decltype(ircd::m::vm::notify_hook) ircd::m::vm::notify_hook { { "name", "vm.notify" } }; ircd::m::vm::eval::eval(const vm::opts &opts) :opts { &opts } ,txn { *dbs::events, db::txn::opts { opts.reserve_bytes + opts.reserve_index, // reserve_bytes 0, // max_bytes (no max) } } { } ircd::m::vm::eval::eval(const event &event, const vm::opts &opts) :eval{opts} { operator()(event); } enum ircd::m::vm::fault ircd::m::vm::eval::operator()(const event &event) try { assert(opts); const event::conforms &report { opts->conforming && !opts->conformed? event::conforms{event, opts->non_conform.report}: opts->report }; if(opts->conforming && !report.clean()) throw error { fault::INVALID, "Non-conforming event: %s", string(report) }; // A conforming (with lots of masks) event without an event_id is an EDU. const fault ret { json::get<"event_id"_>(event)? _eval_pdu(*this, event): _eval_edu(*this, event) }; if(ret != fault::ACCEPT) return ret; vm::accepted accepted { event, opts, &report }; if(opts->notify) { notify_hook(event); vm::accept(accepted); } if(opts->effects) _tmp_effects(event); if(opts->debuglog_accept) log.debug("%s", pretty_oneline(event)); if(opts->infolog_accept) log.info("%s", pretty_oneline(event)); return ret; } catch(const error &e) { if(opts->errorlog & e.code) log.error("eval %s: %s %s", json::get<"event_id"_>(event)?: json::string{""}, e.what(), e.content); if(opts->warnlog & e.code) log.warning("eval %s: %s %s", json::get<"event_id"_>(event)?: json::string{""}, e.what(), e.content); if(opts->nothrows & e.code) return e.code; throw; } catch(const ctx::interrupted &e) { if(opts->errorlog & fault::INTERRUPT) log.error("eval %s: #NMI: %s", json::get<"event_id"_>(event)?: json::string{""}, e.what()); if(opts->warnlog & fault::INTERRUPT) log.warning("eval %s: #NMI: %s", json::get<"event_id"_>(event)?: json::string{""}, e.what()); throw; } catch(const std::exception &e) { if(opts->errorlog & fault::GENERAL) log.error("eval %s: #GP: %s", json::get<"event_id"_>(event)?: json::string{""}, e.what()); if(opts->warnlog & fault::GENERAL) log.warning("eval %s: #GP: %s", json::get<"event_id"_>(event)?: json::string{""}, e.what()); if(opts->nothrows & fault::GENERAL) return fault::GENERAL; throw error { fault::GENERAL, "%s", e.what() }; } enum ircd::m::vm::fault ircd::m::vm::_eval_edu(eval &eval, const event &event) { eval_hook(event); return fault::ACCEPT; } enum ircd::m::vm::fault ircd::m::vm::_eval_pdu(eval &eval, const event &event) { assert(eval.opts); const auto &opts { *eval.opts }; const m::event::id &event_id { at<"event_id"_>(event) }; const m::room::id &room_id { at<"room_id"_>(event) }; if(!opts.replays && exists(event_id)) //TODO: exclusivity throw error { fault::EXISTS, "Event has already been evaluated." }; if(opts.verify) if(!verify(event)) throw m::BAD_SIGNATURE { "Signature verification failed" }; // Obtain sequence number here const uint64_t sequence_number { ++vm::current_sequence }; db::txn::append { eval.txn, dbs::event_seq, { db::op::SET, byte_view(sequence_number), event_id } }; eval_hook(event); const auto &depth { json::get<"depth"_>(event) }; const auto &type { unquote(at<"type"_>(event)) }; const event::prev prev { event }; const size_t prev_count { size(json::get<"prev_events"_>(prev)) }; //TODO: ex if(opts.write && prev_count) { for(size_t i(0); i < prev_count; ++i) { const auto prev_id{prev.prev_event(i)}; if(opts.prev_check_exists && !exists(prev_id)) throw error { fault::EVENT, "Missing prev event %s", string_view{prev_id} }; } int64_t top; id::event::buf head; std::tie(top, head) = m::top(std::nothrow, room_id); if(top < 0 && (opts.head_must_exist || opts.history)) throw error { fault::STATE, "Found nothing for room %s", string_view{room_id} }; m::room room{room_id, head}; m::room::state state{room}; m::state::id_buffer new_root_buf; m::dbs::write_opts wopts; wopts.root_in = state.root_id; wopts.root_out = new_root_buf; wopts.present = opts.present; wopts.history = opts.history; const auto new_root { dbs::write(eval.txn, event, wopts) }; } else if(opts.write) { m::state::id_buffer new_root_buf; m::dbs::write_opts wopts; wopts.root_out = new_root_buf; wopts.present = opts.present; wopts.history = opts.history; const auto new_root { dbs::write(eval.txn, event, wopts) }; } if(opts.write) write(eval); return fault::ACCEPT; } void ircd::m::vm::write(eval &eval) { if(eval.opts->debuglog_accept) log.debug("Committing %zu cells in %zu bytes to events database...", eval.txn.size(), eval.txn.bytes()); eval.txn(); } bool ircd::m::vm::events::rfor_each(const uint64_t &start, const closure_bool &closure) { event::fetch event; return rfor_each(start, id_closure_bool{[&event, &closure] (const uint64_t &seq, const event::id &event_id) { if(!seek(event, event_id, std::nothrow)) return true; return closure(seq, event); }}); } bool ircd::m::vm::events::rfor_each(const uint64_t &start, const id_closure_bool &closure) { auto &column { dbs::event_seq }; if(start == uint64_t(-1)) { for(auto it(column.rbegin()); it; ++it) if(!closure(byte_view(it->first), it->second)) return false; return true; } auto it { column.lower_bound(byte_view(start)) }; for(; it; ++it) if(!closure(byte_view(it->first), it->second)) return false; return true; } bool ircd::m::vm::events::for_each(const uint64_t &start, const closure_bool &closure) { event::fetch event; return for_each(start, id_closure_bool{[&event, &closure] (const uint64_t &seq, const event::id &event_id) { if(!seek(event, event_id, std::nothrow)) return true; return closure(seq, event); }}); } bool ircd::m::vm::events::for_each(const uint64_t &start, const id_closure_bool &closure) { auto &column { dbs::event_seq }; auto it { column.lower_bound(byte_view(start)) }; for(; it; ++it) if(!closure(byte_view(it->first), it->second)) return false; return true; } uint64_t ircd::m::vm::sequence(const eval &eval) { uint64_t ret; eval.txn.at(db::op::SET, "_event_seq", [&ret] (const auto &delta) { const byte_view seqnum { std::get(delta) }; ret = seqnum; }); return ret; } uint64_t ircd::m::vm::last_sequence() { id::event::buf event_id; return last_sequence(event_id); } uint64_t ircd::m::vm::last_sequence(id::event::buf &event_id) { auto &column { dbs::event_seq }; const auto it { column.rbegin() }; if(!it) { // If this iterator is invalid the events db should // be completely fresh. assert(db::sequence(*dbs::events) == 0); return 0; } event_id = it->second; return byte_view(it->first); } ircd::string_view ircd::m::vm::reflect(const enum fault &code) { switch(code) { case fault::ACCEPT: return "ACCEPT"; case fault::EXISTS: return "EXISTS"; case fault::INVALID: return "INVALID"; case fault::DEBUGSTEP: return "DEBUGSTEP"; case fault::BREAKPOINT: return "BREAKPOINT"; case fault::GENERAL: return "GENERAL"; case fault::EVENT: return "EVENT"; case fault::STATE: return "STATE"; case fault::INTERRUPT: return "INTERRUPT"; } return "??????"; } //TODO: X void ircd::m::vm::_tmp_effects(const m::event &event) { const auto &type{at<"type"_>(event)}; //TODO: X if(type == "m.room.join_rules") { const m::room::id room_id{at<"room_id"_>(event)}; const m::user::id sender{at<"sender"_>(event)}; if(my_host(sender.host())) send(room::id{"!public:zemos.net"}, sender, "ircd.room", room_id, {}); } //TODO: X if(type == "m.room.create") { const string_view local{m::room::id{at<"room_id"_>(event)}.localname()}; if(local != "users") //TODO: circ dep send(my_room, at<"sender"_>(event), "ircd.room", at<"room_id"_>(event), json::object{}); } } // // accepted // ircd::m::vm::accepted::accepted(const m::event &event, const vm::opts *const &opts, const event::conforms *const &report) :m::event{event} ,context{ctx::current} ,opts{opts} ,report{report} { }