mirror of
https://github.com/matrix-construct/construct
synced 2024-09-28 19:58:53 +02:00
ircd:Ⓜ️:feds: Add / upgrade backfill.
This commit is contained in:
Jason Volk
parent
7ca0cd5e34
commit
eace5e309c
4 changed files with 96 additions and 189 deletions
|
|
@ -33,6 +33,7 @@ namespace ircd::m::feds
|
|||
bool head(const opts &, const closure &);
|
||||
bool event(const opts &, const closure &);
|
||||
bool state(const opts &, const closure &);
|
||||
bool backfill(const opts &, const closure &);
|
||||
bool version(const opts &, const closure &);
|
||||
bool perspective(const opts &, const closure &);
|
||||
};
|
||||
|
|
@ -52,4 +53,5 @@ struct ircd::m::feds::opts
|
|||
m::event::id event_id;
|
||||
m::user::id user_id;
|
||||
string_view arg[4]; // misc argv
|
||||
uint64_t argi[4]; // misc integer argv
|
||||
};
|
||||
|
|
|
|||
14
ircd/m.cc
14
ircd/m.cc
|
|
@ -1087,6 +1087,20 @@ ircd::m::feds::state(const opts &o,
|
|||
return call(o, c);
|
||||
}
|
||||
|
||||
bool
|
||||
ircd::m::feds::backfill(const opts &o,
|
||||
const closure &c)
|
||||
{
|
||||
using prototype = bool (const opts &, const closure &);
|
||||
|
||||
static mods::import<prototype> call
|
||||
{
|
||||
"federation_federation", "ircd::m::feds::backfill"
|
||||
};
|
||||
|
||||
return call(o, c);
|
||||
}
|
||||
|
||||
bool
|
||||
ircd::m::feds::event(const opts &o,
|
||||
const closure &c)
|
||||
|
|
|
|||
|
|
@ -10451,17 +10451,58 @@ console_cmd__feds__backfill(opt &out, const string_view &line)
|
|||
param.at(2, size_t(4))
|
||||
};
|
||||
|
||||
using prototype = void (const m::room::id &,
|
||||
const m::event::id &,
|
||||
const size_t &,
|
||||
std::ostream &);
|
||||
std::map<std::string, std::set<std::string>, std::less<>> grid;
|
||||
std::set<string_view> origins;
|
||||
|
||||
static mods::import<prototype> feds__backfill
|
||||
m::feds::opts opts;
|
||||
opts.room_id = room_id;
|
||||
opts.event_id = event_id;
|
||||
opts.argi[0] = limit;
|
||||
m::feds::backfill(opts, [&grid, &origins]
|
||||
(const auto &result)
|
||||
{
|
||||
"federation_federation", "feds__backfill"
|
||||
};
|
||||
if(result.eptr)
|
||||
return true;
|
||||
|
||||
const json::array &pdus
|
||||
{
|
||||
result.object["pdus"]
|
||||
};
|
||||
|
||||
for(const json::object &pdu : pdus)
|
||||
{
|
||||
const auto &event_id
|
||||
{
|
||||
unquote(pdu.at("event_id"))
|
||||
};
|
||||
|
||||
auto it(grid.lower_bound(event_id));
|
||||
if(it == end(grid) || it->first != event_id)
|
||||
it = grid.emplace_hint(it, event_id, std::set<std::string>{});
|
||||
|
||||
auto &set(it->second);
|
||||
const auto iit(set.emplace(result.origin));
|
||||
origins.emplace(*iit.first);
|
||||
}
|
||||
|
||||
return true;
|
||||
});
|
||||
|
||||
size_t i(0);
|
||||
for(const auto &p : grid)
|
||||
out << i++ << " " << p.first << std::endl;
|
||||
|
||||
for(size_t j(0); j < i; ++j)
|
||||
out << "| " << std::left << std::setw(2) << j;
|
||||
out << "|" << std::endl;
|
||||
|
||||
for(const auto &origin : origins)
|
||||
{
|
||||
for(const auto &p : grid)
|
||||
out << "| " << (p.second.count(origin)? '+' : ' ') << " ";
|
||||
out << "| " << origin << std::endl;
|
||||
}
|
||||
|
||||
feds__backfill(room_id, event_id, limit, out);
|
||||
return true;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -113,6 +113,37 @@ ircd::m::feds::version(const opts &opts,
|
|||
return handler(opts, closure, requests);
|
||||
}
|
||||
|
||||
bool
|
||||
IRCD_MODULE_EXPORT
|
||||
ircd::m::feds::backfill(const opts &opts,
|
||||
const closure &closure)
|
||||
{
|
||||
const auto make_request{[&opts]
|
||||
(auto &request, const auto &origin)
|
||||
{
|
||||
m::v1::backfill::opts v1opts;
|
||||
v1opts.dynamic = true;
|
||||
v1opts.event_id = opts.event_id;
|
||||
v1opts.limit = opts.argi[0];
|
||||
v1opts.remote = string_view
|
||||
{
|
||||
request.origin, strlcpy{request.origin, origin}
|
||||
};
|
||||
|
||||
return m::v1::backfill
|
||||
{
|
||||
opts.room_id, request.buf, std::move(v1opts)
|
||||
};
|
||||
}};
|
||||
|
||||
auto requests
|
||||
{
|
||||
creator<m::v1::backfill>(opts, make_request)
|
||||
};
|
||||
|
||||
return handler(opts, closure, requests);
|
||||
}
|
||||
|
||||
bool
|
||||
IRCD_MODULE_EXPORT
|
||||
ircd::m::feds::state(const opts &opts,
|
||||
|
|
@ -284,184 +315,3 @@ ircd::m::feds::handler(const opts &opts,
|
|||
|
||||
return true;
|
||||
}
|
||||
|
||||
//
|
||||
// legacy
|
||||
//
|
||||
|
||||
extern "C" void
|
||||
feds__event(const m::event::id &event_id, std::ostream &out)
|
||||
{
|
||||
const m::room::id::buf room_id{[&event_id]
|
||||
{
|
||||
const m::event::fetch event
|
||||
{
|
||||
event_id
|
||||
};
|
||||
|
||||
return m::room::id::buf{at<"room_id"_>(event)};
|
||||
}()};
|
||||
|
||||
struct req
|
||||
:m::v1::event
|
||||
{
|
||||
char origin[256];
|
||||
char buf[96_KiB];
|
||||
|
||||
req(const m::event::id &event_id, const string_view &origin)
|
||||
:m::v1::event{[&]
|
||||
{
|
||||
m::v1::event::opts opts;
|
||||
opts.dynamic = false;
|
||||
opts.remote = string_view{this->origin, strlcpy(this->origin, origin)};
|
||||
return m::v1::event{event_id, mutable_buffer{buf}, std::move(opts)};
|
||||
}()}
|
||||
{}
|
||||
};
|
||||
|
||||
std::list<req> reqs;
|
||||
const m::room::origins origins{room_id};
|
||||
origins.for_each([&out, &event_id, &reqs]
|
||||
(const string_view &origin)
|
||||
{
|
||||
const auto emsg
|
||||
{
|
||||
ircd::server::errmsg(origin)
|
||||
};
|
||||
|
||||
if(emsg)
|
||||
{
|
||||
out << "! " << origin << " " << emsg << std::endl;
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
reqs.emplace_back(event_id, origin);
|
||||
}
|
||||
catch(const std::exception &e)
|
||||
{
|
||||
out << "! " << origin << " " << e.what() << std::endl;
|
||||
return;
|
||||
}
|
||||
});
|
||||
|
||||
auto all
|
||||
{
|
||||
ctx::when_all(begin(reqs), end(reqs))
|
||||
};
|
||||
|
||||
all.wait(30s, std::nothrow);
|
||||
|
||||
for(auto &req : reqs) try
|
||||
{
|
||||
if(req.wait(1ms, std::nothrow))
|
||||
{
|
||||
const auto code{req.get()};
|
||||
out << "+ " << req.origin << " " << http::status(code) << std::endl;
|
||||
}
|
||||
else cancel(req);
|
||||
}
|
||||
catch(const std::exception &e)
|
||||
{
|
||||
out << "- " << req.origin << " " << e.what() << std::endl;
|
||||
}
|
||||
}
|
||||
|
||||
extern "C" void
|
||||
feds__backfill(const m::room::id &room_id,
|
||||
const m::event::id &event_id,
|
||||
const size_t &limit,
|
||||
std::ostream &out)
|
||||
{
|
||||
struct req
|
||||
:m::v1::backfill
|
||||
{
|
||||
char origin[256];
|
||||
char buf[16_KiB];
|
||||
|
||||
req(const m::room::id &room_id, const m::event::id &event_id, const string_view &origin, const size_t &limit)
|
||||
:m::v1::backfill{[&]
|
||||
{
|
||||
m::v1::backfill::opts opts;
|
||||
opts.remote = string_view{this->origin, strlcpy(this->origin, origin)};
|
||||
opts.event_id = event_id;
|
||||
opts.limit = limit;
|
||||
opts.dynamic = true;
|
||||
return m::v1::backfill{room_id, mutable_buffer{buf}, std::move(opts)};
|
||||
}()}
|
||||
{}
|
||||
};
|
||||
|
||||
std::list<req> reqs;
|
||||
const m::room::origins origins{room_id};
|
||||
origins.for_each([&out, &room_id, &event_id, &limit, &reqs]
|
||||
(const string_view &origin)
|
||||
{
|
||||
const auto emsg
|
||||
{
|
||||
ircd::server::errmsg(origin)
|
||||
};
|
||||
|
||||
if(!emsg) try
|
||||
{
|
||||
reqs.emplace_back(room_id, event_id, origin, limit);
|
||||
}
|
||||
catch(const std::exception &e)
|
||||
{
|
||||
return;
|
||||
}
|
||||
});
|
||||
|
||||
auto all
|
||||
{
|
||||
ctx::when_all(begin(reqs), end(reqs))
|
||||
};
|
||||
|
||||
all.wait(30s, std::nothrow);
|
||||
|
||||
std::map<string_view, std::set<const req *>> grid;
|
||||
|
||||
for(auto &req : reqs) try
|
||||
{
|
||||
if(req.wait(1ms, std::nothrow))
|
||||
{
|
||||
const auto code{req.get()};
|
||||
const json::object &response{req};
|
||||
const json::array pdus
|
||||
{
|
||||
response.at("pdus")
|
||||
};
|
||||
|
||||
for(const json::object &pdu : pdus)
|
||||
{
|
||||
const auto &event_id
|
||||
{
|
||||
unquote(pdu.at("event_id"))
|
||||
};
|
||||
|
||||
grid[event_id].emplace(&req);
|
||||
}
|
||||
}
|
||||
else cancel(req);
|
||||
}
|
||||
catch(const std::exception &e)
|
||||
{
|
||||
out << "- " << req.origin << " " << e.what() << std::endl;
|
||||
}
|
||||
|
||||
size_t i(0);
|
||||
for(const auto &p : grid)
|
||||
out << i++ << " " << p.first << std::endl;
|
||||
|
||||
for(size_t j(0); j < i; ++j)
|
||||
out << "| " << std::left << std::setw(2) << j;
|
||||
out << "|" << std::endl;
|
||||
|
||||
for(const auto &req : reqs)
|
||||
{
|
||||
for(const auto &p : grid)
|
||||
out << "| " << (p.second.count(&req)? '+' : ' ') << " ";
|
||||
out << "| " << req.origin << std::endl;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue