/* vim: set expandtab ts=4 sw=4: */
/*
* You may redistribute this program and/or modify it under the terms of
* the GNU General Public License as published by the Free Software Foundation,
* either version 3 of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see .
*/
#include "subnode/ReachabilityAnnouncer.h"
#include "subnode/PeeringSeeder.h"
#include "util/events/Timeout.h"
#include "util/Identity.h"
#include "util/events/Time.h"
#include "wire/Announce.h"
#include "crypto/AddressCalc.h"
#include "crypto/Sign.h"
#include "util/AddrTools.h"
#include "util/Hex.h"
#include "util/Hash.h"
#include "rust/cjdns_sys/Rffi.h"
#include "util/Defined.h"
#include "benc/Dict.h"
#include
// This is the time between the timestamp of the newest message and the point where
// snode and subnode agree to drop messages from the snode state.
#define AGREED_TIMEOUT_MS (1000 * 60 * 20)
#define MSG_SIZE_LIMIT 700
// Initial time between messages is 60 seconds, adjusted based on amount of full messages
#define INITIAL_TBA 60000
#define ArrayList_TYPE Message_t
#define ArrayList_NAME OfMessages
#include "util/ArrayList.h"
#define ArrayList_TYPE struct Announce_ItemHeader
#define ArrayList_NAME OfAnnItems
#include "util/ArrayList.h"
#define ArrayList_TYPE struct Announce_Peer
#define ArrayList_NAME OfBarePeers
#include "util/ArrayList.h"
// -- Generic Functions -- //
// We must reannounce before the agreed timeout because if it happens that there are
// too many peers to fit in one packet, the packet will go out and re-announce the ones
// who fit but the others will not fit in the packet and once the timestamp comes in,
// they will be pulled by the route server.
//
// We could just declare that we are re-announcing everything at minute 15 but if we
// do so then there will potentially be be a flood of full packets every 15 minutes
// and link state will not be communicated.
//
// To fix this, we begin re-announcing after 14 minutes, which peers are eligable to be
// re-announced is randomized by the timestamp of the previous announcement (something
// which changes each cycle). Re-announcements occur between minutes 14 and minutes 19
// with the last minute reserved as a 1 minute "quiet period" where announcements can
// catch up before minute 20 when peers will be dropped by the route server.
//
#define QUIET_PERIOD_MS (1000 * 60)
static int64_t timeUntilReannounce(
int64_t nowServerTime,
int64_t lastAnnouncedTime,
struct Announce_ItemHeader* item)
{
uint32_t hash = Hash_compute((uint8_t*)item, item->length);
int64_t timeSince = nowServerTime - lastAnnouncedTime;
int64_t random5Min = (((uint64_t)lastAnnouncedTime + hash) % 600) * 1000;
return (AGREED_TIMEOUT_MS - QUIET_PERIOD_MS) - (timeSince + random5Min);
}
static int64_t timestampFromMsg(Message_t* msg)
{
struct Announce_Header* hdr = (struct Announce_Header*) Message_bytes(msg);
Assert_true(Message_getLength(msg) >= Announce_Header_SIZE);
return Announce_Header_getTimestamp(hdr);
}
static struct Announce_ItemHeader* itemFromSnodeState(struct ArrayList_OfMessages* snodeState,
struct Announce_ItemHeader* ref,
int64_t sinceTime,
int64_t* timeOut)
{
for (int i = snodeState->length - 1; i >= 0; i--) {
Message_t* msg = ArrayList_OfMessages_get(snodeState, i);
struct Announce_ItemHeader* item = Announce_itemInMessage(msg, ref);
if (!item) { continue; }
int64_t ts = timestampFromMsg(msg);
if (sinceTime > ts) { return NULL; }
if (timeOut) { *timeOut = ts; }
return item;
}
return NULL;
}
// Calculate the sha512 of a message list where a given set of signed messages will corrispond
// to a given hash.
static void hashMsgList(struct ArrayList_OfMessages* msgList, uint8_t out[64])
{
uint8_t hash[64] = {0};
for (int i = 0; i < msgList->length; i++) {
Message_t* msg = ArrayList_OfMessages_get(msgList, i);
Err_assert(Message_epush(msg, hash, 64));
Rffi_crypto_hash_sha512(hash, Message_bytes(msg), Message_getLength(msg));
Err_assert(Message_epop(msg, NULL, 64));
}
Bits_memcpy(out, hash, 64);
}
static int64_t estimateClockSkew(int64_t sentTime, int64_t snodeRecvTime, int64_t now)
{
// We estimate that the snode received our message at time: 1/2 the RTT
int64_t halfRtt = sentTime + ((now - sentTime) / 2);
return halfRtt - snodeRecvTime;
}
// We'll try to halve our estimated clock skew each RTT so on average it should eventually
// target in on the exact skew. Ideal would be to use a rolling average such that one
// screwy RTT has little effect but that's more work.
static int64_t estimateImprovedClockSkew(int64_t sentTime,
int64_t snodeRecvTime,
int64_t now,
int64_t lastSkew)
{
int64_t thisSkew = estimateClockSkew(sentTime, snodeRecvTime, now);
int64_t skewDiff = thisSkew - lastSkew;
return lastSkew + (skewDiff / 2);
}
// -- Context -- //
// Depending on what news we have learned, we will adopt one of a set of possible states
// whcih inform how often we contact our supernode. The numeric representation of the
// state corrisponds to the number of milliseconds between messages to be sent to our
// supernode.
enum ReachabilityAnnouncer_State
{
// The message we build up from our local state is full, we obviously need to send it
// asap in order that we can finish informing the snode of our peers.
ReachabilityAnnouncer_State_MSGFULL = 500,
// In this state we know how to reach the snode but we have no announced reachability
// (so we are effectively offline) we have to announce quickly in order to be online.
ReachabilityAnnouncer_State_FIRSTPEER = 1000,
// We have just dropped a peer, we should announce quickly in order to help the snode
// know that our link is dead.
ReachabilityAnnouncer_State_PEERGONE = 6000,
// We have picked up a new peer, we should announce moderately fast in order to make
// sure that the snode picks the best path out of the possible options.
ReachabilityAnnouncer_State_NEWPEER = 12000,
// No new peers or dropped peers, we'll just send announcements at a low interval in
// order to keep our snode up to date on latencies and drop percentages of different
// links. Cadence is regulated by timeBetweenAnns.
ReachabilityAnnouncer_State_NORMAL = -1
};
static const char* printState(enum ReachabilityAnnouncer_State s)
{
switch (s) {
case ReachabilityAnnouncer_State_FIRSTPEER: return "FIRSTPEER";
case ReachabilityAnnouncer_State_PEERGONE: return "PEERGONE";
case ReachabilityAnnouncer_State_NEWPEER: return "NEWPEER";
case ReachabilityAnnouncer_State_NORMAL: return "NORMAL";
default: return "unknown";
}
}
struct ReachabilityAnnouncer_pvt;
struct Query {
struct ReachabilityAnnouncer_pvt* rap;
Message_t* msg;
struct Address target;
Identity
};
struct ReachabilityAnnouncer_pvt
{
struct ReachabilityAnnouncer pub;
struct Timeout* announceCycle;
struct Allocator* alloc;
struct Log* log;
EventBase_t* base;
struct MsgCore* msgCore;
struct Random* rand;
struct SupernodeHunter* snh;
struct EncodingScheme* myScheme;
struct ReachabilityCollector* rc;
PeeringSeeder_t* ps;
String* encodingSchemeStr;
struct Announce_ItemHeader* mySchemeItem;
uint8_t signingKeypair[64];
uint8_t pubSigningKey[32];
int64_t timeOfLastReply;
// The cjdns clock is monotonic and is calibrated once on launch so clockSkew
// will be reliable even if the machine also has NTP and NTP also changes the clock
// clockSkew is literally the number of milliseconds which we believe our clock is ahead of
// our supernode's clock.
int64_t clockSkew;
struct Address snode;
// This is effectively a log which means we add messages to it as time goes but we remove
// messages which are more than AGREED_TIMEOUT_MS (20 minutes) older than the most recent
// message in the list (the one at the highest index). We also identify messages in the list
// which update only peers that have been updated again since and we remove those as well.
// IMPORTANT: The removal of messages from this list is using the same algorithm that is used
// on the supernode and if it changes then they will desync and go into a reset
// loop.
struct ArrayList_OfMessages* snodeState;
struct Query* onTheWire;
// this is by our clock, not skewed to the snode time.
int64_t msgOnWireSentTime;
// If true then when we send nextMsg, it will be a state reset of the node.
bool resetState;
enum ReachabilityAnnouncer_State state;
int timeBetweenAnns;
Identity
};
// -- "Methods" -- //
static int64_t ourTime(struct ReachabilityAnnouncer_pvt* rap)
{
uint64_t now = Time_currentTimeMilliseconds();
Assert_true(!(now >> 63));
return (int64_t) now;
}
static int64_t snTime(struct ReachabilityAnnouncer_pvt* rap)
{
return ourTime(rap) - rap->clockSkew;
}
static char* printPeer(
char out[60],
struct ReachabilityAnnouncer_pvt* rap,
struct Announce_Peer* p)
{
uint64_t path = Endian_bigEndianToHost16(p->peerNum_be);
AddrTools_printPath(out, path);
out[19] = '.';
AddrTools_printIp(&out[20], p->peerIpv6);
return out;
}
static char* printItem(
char out[60],
struct ReachabilityAnnouncer_pvt* rap,
struct Announce_ItemHeader* item)
{
if (item->type == Announce_Type_PEER) {
struct Announce_Peer* p = (struct Announce_Peer*) item;
return printPeer(out, rap, p);
} else if (item->type == Announce_Type_ENCODING_SCHEME) {
return "encoding scheme";
} else if (item->type == Announce_Type_VERSION) {
return "version";
} else {
return "unknown";
}
}
static bool pushLinkState(struct ReachabilityAnnouncer_pvt* rap,
Message_t* msg)
{
for (int i = 0;; i++) {
struct ReachabilityCollector_PeerInfo* pi = ReachabilityCollector_getPeerInfo(rap->rc, i);
if (!pi || !pi->pathThemToUs) { break; }
int lastLen = Message_getLength(msg);
pi->linkState.nodeId = pi->addr.path & 0xffff;
if (LinkState_encode(msg, &pi->linkState, pi->lastAnnouncedSamples)) {
Log_debug(rap->log, "Failed to add link state for [%s]",
Address_toString(&pi->addr, Message_getAlloc(msg))->bytes);
}
if (Message_getLength(msg) > MSG_SIZE_LIMIT) {
Err_assert(Message_epop(msg, NULL, Message_getLength(msg) - lastLen));
Log_debug(rap->log, "Couldn't add link state for [%s] (out of space)",
Address_toString(&pi->addr, Message_getAlloc(msg))->bytes);
return true;
} else {
Log_debug(rap->log, "Updated link state for [%s]",
Address_toString(&pi->addr, Message_getAlloc(msg))->bytes);
pi->lastAnnouncedSamples = pi->linkState.samples;
}
}
return false;
}
// Insert or update the state information for a peer in a msgList
#define updateItem_NOOP 0
#define updateItem_ADD 1
#define updateItem_UPDATE 2
#define updateItem_ENOSPACE -1
static int updateItem(struct ReachabilityAnnouncer_pvt* rap,
Message_t* msg,
struct Announce_ItemHeader* refItem)
{
char buf[60];
const char* logInfo = "";
if (Defined(Log_DEBUG)) {
logInfo = printItem(buf, rap, refItem);
}
int64_t serverTime = snTime(rap);
int64_t sinceTime = serverTime - AGREED_TIMEOUT_MS;
struct Announce_ItemHeader* item = NULL;
if (rap->onTheWire) {
item = Announce_itemInMessage(rap->onTheWire->msg, refItem);
}
if (!item) {
int64_t peerTime = 0;
item = itemFromSnodeState(rap->snodeState, refItem, sinceTime, &peerTime);
if (item && Announce_ItemHeader_equals(item, refItem)) {
int64_t tur = timeUntilReannounce(serverTime, peerTime, item);
if (tur < 0) {
Log_debug(rap->log, "updateItem [%s] needs re-announce", logInfo);
} else {
Log_debug(rap->log, "updateItem [%s] no re-announce for [%d] sec",
logInfo, (int)(tur / 1000));
return updateItem_NOOP;
}
} else if (item) {
Log_debug(rap->log, "updateItem [%s] needs update (changed)", logInfo);
} else {
Log_debug(rap->log, "updateItem [%s] not found in snodeState", logInfo);
}
} else if (Announce_ItemHeader_equals(item, refItem)) {
Log_debug(rap->log, "updateItem [%s] found onTheWire, noop", logInfo);
return updateItem_NOOP;
} else {
Log_debug(rap->log, "updateItem [%s] found onTheWire but needs update", logInfo);
}
if (Message_getLength(msg) > MSG_SIZE_LIMIT) {
Log_debug(rap->log, "updateItem [%s] msg is too big to [%s] item",
logInfo, item ? "UPDATE" : "INSERT");
return updateItem_ENOSPACE;
}
Err_assert(Message_epush(msg, refItem, refItem->length));
while ((uintptr_t)Message_bytes(msg) % 4) {
// Ensure alignment
Err_assert(Message_epush8(msg, 1));
}
return (item) ? updateItem_UPDATE : updateItem_ADD;
}
static void stateUpdate(struct ReachabilityAnnouncer_pvt* rap, enum ReachabilityAnnouncer_State st)
{
if (rap->state < st) { return; }
rap->state = st;
}
static void annPeerForPi(struct ReachabilityAnnouncer_pvt* rap,
struct Announce_Peer* apOut,
struct ReachabilityCollector_PeerInfo* pi)
{
Assert_true(pi);
Announce_Peer_init(apOut);
apOut->encodingFormNum = EncodingScheme_getFormNum(rap->myScheme, pi->addr.path);
apOut->peerNum_be = Endian_hostToBigEndian16(pi->addr.path & 0xffff);
Bits_memcpy(apOut->peerIpv6, pi->addr.ip6.bytes, 16);
apOut->label_be = Endian_hostToBigEndian32(pi->pathThemToUs);
}
static bool pushPeers(struct ReachabilityAnnouncer_pvt* rap, Message_t* msg)
{
for (int i = 0;; i++) {
struct ReachabilityCollector_PeerInfo* pi = ReachabilityCollector_getPeerInfo(rap->rc, i);
if (!pi || !pi->pathThemToUs) { return false; }
struct Announce_Peer annP;
annPeerForPi(rap, &annP, pi);
if (updateItem(rap, msg, (struct Announce_ItemHeader*) &annP) == updateItem_ENOSPACE) {
return true;
}
}
}
static void stateReset(struct ReachabilityAnnouncer_pvt* rap)
{
for (int i = rap->snodeState->length - 1; i >= 0; i--) {
Message_t* msg = ArrayList_OfMessages_remove(rap->snodeState, i);
Allocator_free(Message_getAlloc(msg));
}
if (rap->onTheWire) {
// this message is owned by a ping allocator so it will be freed by that
rap->onTheWire = NULL;
}
// we must force the state to FIRSTPEER
rap->state = ReachabilityAnnouncer_State_FIRSTPEER;
rap->timeBetweenAnns = INITIAL_TBA;
rap->resetState = true;
}
static void addServerStateMsg(struct ReachabilityAnnouncer_pvt* rap, Message_t* msg)
{
Assert_true(Message_getLength(msg) >= Announce_Header_SIZE);
int64_t mostRecentTime = timestampFromMsg(msg);
int64_t sinceTime = mostRecentTime - AGREED_TIMEOUT_MS;
ArrayList_OfMessages_add(rap->snodeState, msg);
// Filter completely redundant messages and messages older than sinceTime
struct Allocator* tempAlloc = Allocator_child(rap->alloc);
struct ArrayList_OfAnnItems* knownItems = ArrayList_OfAnnItems_new(tempAlloc);
for (int i = rap->snodeState->length - 1; i >= 0; i--) {
bool redundant = true;
Message_t* m = ArrayList_OfMessages_get(rap->snodeState, i);
struct Announce_ItemHeader* item = Announce_ItemHeader_next(m, NULL);
for (; item; item = Announce_ItemHeader_next(m, item)) {
if (Announce_ItemHeader_isEphimeral(item)) {
// Ephimeral items do not make a message non-redundant
continue;
}
bool inList = false;
for (int j = 0; j < knownItems->length; j++) {
struct Announce_ItemHeader* knownItem = ArrayList_OfAnnItems_get(knownItems, j);
if (Announce_ItemHeader_doesReplace(knownItem, item)) {
inList = true;
break;
}
}
if (!inList) {
ArrayList_OfAnnItems_add(knownItems, item);
redundant = false;
}
}
if (redundant && m != msg) {
ArrayList_OfMessages_remove(rap->snodeState, i);
Allocator_free(Message_getAlloc(m));
} else if (timestampFromMsg(m) < sinceTime) {
// this will cause an immediate reset of state because we don't remove it and
// the server side will.
Log_warn(rap->log, "Announcement expiring which has not been replaced in time");
}
}
Allocator_free(tempAlloc);
}
static struct ArrayList_OfBarePeers* getSnodeStatePeers(
struct ReachabilityAnnouncer_pvt* rap,
struct Allocator* alloc)
{
struct ArrayList_OfBarePeers* out = ArrayList_OfBarePeers_new(alloc);
for (int i = 0; i < rap->snodeState->length; i++) {
Message_t* snm = ArrayList_OfMessages_get(rap->snodeState, i);
struct Announce_Peer* p = NULL;
for (p = Announce_Peer_next(snm, NULL); p; p = Announce_Peer_next(snm, p)) {
bool found = false;
for (int j = 0; j < out->length; j++) {
struct Announce_Peer* p1 = ArrayList_OfBarePeers_get(out, j);
if (p1->peerNum_be == p->peerNum_be) {
Bits_memcpy(p1, p, sizeof(struct Announce_Peer));
found = true;
}
}
if (!found) {
struct Announce_Peer* p1 = Allocator_clone(alloc, p);
ArrayList_OfBarePeers_add(out, p1);
}
}
}
for (int j = out->length - 1; j >= 0; j--) {
struct Announce_Peer* p1 = ArrayList_OfBarePeers_get(out, j);
if (!p1->label_be) { ArrayList_OfBarePeers_remove(out, j); }
}
return out;
}
// -- Public -- //
void ReachabilityAnnouncer_updatePeer(struct ReachabilityAnnouncer* ra,
struct Address* nodeAddr,
struct ReachabilityCollector_PeerInfo* pi)
{
struct ReachabilityAnnouncer_pvt* rap = Identity_check((struct ReachabilityAnnouncer_pvt*) ra);
struct Allocator* tempAlloc = Allocator_child(rap->alloc);
if (!pi) {
Log_debug(rap->log, "Update for [%s] - gone", Address_toString(nodeAddr, tempAlloc)->bytes);
stateUpdate(rap, ReachabilityAnnouncer_State_PEERGONE);
} else {
struct ArrayList_OfBarePeers* snodeState = getSnodeStatePeers(rap, tempAlloc);
if (snodeState->length == 0) {
Log_debug(rap->log, "Update for [%s] - first peer",
Address_toString(nodeAddr, tempAlloc)->bytes);
stateUpdate(rap, ReachabilityAnnouncer_State_FIRSTPEER);
} else {
Log_debug(rap->log, "Update for [%s] - new peer",
Address_toString(nodeAddr, tempAlloc)->bytes);
stateUpdate(rap, ReachabilityAnnouncer_State_NEWPEER);
}
}
Allocator_free(tempAlloc);
}
// -- Event Callbacks -- //
static void onReplyTimeout(struct ReachabilityAnnouncer_pvt* rap, struct Query* q)
{
// TODO(cjd): one lost packet shouldn't trigger unreachable state
if (!Bits_memcmp(&q->target, &rap->snode, Address_SIZE)) {
rap->snh->snodeIsReachable = false;
if (rap->snh->onSnodeUnreachable) {
rap->snh->onSnodeUnreachable(rap->snh, 0, 0);
}
}
}
static void onReply(Dict* msg, struct Address* src, struct MsgCore_Promise* prom)
{
struct Query* q = Identity_check((struct Query*) prom->userData);
struct ReachabilityAnnouncer_pvt* rap = Identity_check(q->rap);
if (rap->onTheWire != q) {
Log_debug(rap->log, "Got a reply from [%s] which was outstanding when "
"we triggered a state reset, discarding",
Address_toString(prom->target, prom->alloc)->bytes);
return;
}
rap->onTheWire = NULL;
if (!src) {
onReplyTimeout(rap, q);
return;
}
int64_t* snodeRecvTime = Dict_getIntC(msg, "recvTime");
if (!snodeRecvTime) {
Log_warn(rap->log, "snode did not send back recvTime");
onReplyTimeout(rap, q);
return;
}
int64_t sentTime = rap->msgOnWireSentTime;
Log_debug(rap->log, "snode messages before [%d]", rap->snodeState->length);
// We need to takeover the message allocator because it belongs to the ping message which
// will auto-free at the end of this cycle.
Allocator_adopt(rap->alloc, Message_getAlloc(q->msg));
addServerStateMsg(rap, q->msg);
Log_debug(rap->log, "snode messages after [%d]", rap->snodeState->length);
rap->resetState = false;
int64_t now = rap->timeOfLastReply = ourTime(rap);
int64_t oldClockSkew = rap->clockSkew;
Log_debug(rap->log, "sentTime [%lld]", (long long int) sentTime);
Log_debug(rap->log, "snodeRecvTime [%lld]", (long long int) *snodeRecvTime);
Log_debug(rap->log, "now [%lld]", (long long int) now);
Log_debug(rap->log, "oldClockSkew [%lld]", (long long int) oldClockSkew);
rap->clockSkew = estimateImprovedClockSkew(sentTime, *snodeRecvTime, now, oldClockSkew);
Log_debug(rap->log, "Adjusting clock skew by [%lld]",
(long long int) (rap->clockSkew - oldClockSkew));
Log_debug(rap->log, "State [%s]", printState(rap->state));
Log_debug(rap->log, "TBA [%d]", rap->timeBetweenAnns);
rap->state = ReachabilityAnnouncer_State_NORMAL;
String* snodeStateHash = Dict_getStringC(msg, "stateHash");
uint8_t ourStateHash[64];
hashMsgList(rap->snodeState, ourStateHash);
if (!snodeStateHash) {
Log_warn(rap->log, "no stateHash in reply from snode");
} else if (snodeStateHash->len != 64) {
Log_warn(rap->log, "bad stateHash in reply from snode");
} else if (Bits_memcmp(snodeStateHash->bytes, ourStateHash, 64)) {
uint8_t snodeHash[129];
Assert_true(128 == Hex_encode(snodeHash, 129, snodeStateHash->bytes, 64));
uint8_t ourHash[129];
Assert_true(128 == Hex_encode(ourHash, 129, ourStateHash, 64));
Log_warn(rap->log, "state mismatch with snode, [%u] announces\n[%s]\n[%s]",
rap->snodeState->length, snodeHash, ourHash);
} else {
return;
}
Log_warn(rap->log, "desynchronized with snode, resetting state");
stateReset(rap);
}
static bool pushMeta(struct ReachabilityAnnouncer_pvt* rap, Message_t* msg)
{
struct Announce_Version version;
Announce_Version_init(&version);
if (updateItem(rap, msg, (struct Announce_ItemHeader*)&version) == updateItem_ENOSPACE) {
return true;
} else if (updateItem(rap, msg, rap->mySchemeItem) == updateItem_ENOSPACE) {
return true;
}
return false;
}
static bool pushWithdrawLinks(struct ReachabilityAnnouncer_pvt* rap, Message_t* msg)
{
// First withdraw any announcements which are nolonger valid
struct Allocator* tempAlloc = Allocator_child(rap->alloc);
struct ArrayList_OfBarePeers* snodePeers = getSnodeStatePeers(rap, tempAlloc);
bool outOfSpace = false;
for (int i = 0; i < snodePeers->length; i++) {
struct Announce_Peer* p = ArrayList_OfBarePeers_get(snodePeers, i);
uint64_t path = Endian_bigEndianToHost16(p->peerNum_be);
struct ReachabilityCollector_PeerInfo* pi =
ReachabilityCollector_piForLabel(rap->rc, path);
if (pi && pi->pathThemToUs) { continue; }
char buf[60];
Log_debug(rap->log, "Withdrawing route to [%s]", printPeer(buf, rap, p));
p->label_be = 0;
if (updateItem(rap, msg, (struct Announce_ItemHeader*) p) == updateItem_ENOSPACE) {
outOfSpace = true;
break;
}
}
Allocator_free(tempAlloc);
return outOfSpace;
}
static void onAnnounceCycle(void* vRap)
{
struct ReachabilityAnnouncer_pvt* rap =
Identity_check((struct ReachabilityAnnouncer_pvt*) vRap);
// Message out on the wire...
if (rap->onTheWire) { return; }
if (!rap->snode.path) { return; }
int64_t now = ourTime(rap);
int64_t snNow = snTime(rap);
// Not time to send yet?
int64_t delay = now - rap->timeOfLastReply;
if (rap->state == ReachabilityAnnouncer_State_NORMAL) {
if (delay < rap->timeBetweenAnns) { return; }
} else {
if (delay < rap->state) { return; }
}
struct MsgCore_Promise* qp = MsgCore_createQuery(rap->msgCore, 0, rap->alloc);
struct Allocator* queryAlloc = Allocator_child(qp->alloc);
Message_t* msg = Message_new(0, 1300, queryAlloc);
Log_debug(rap->log, "\n");
do {
if (pushMeta(rap, msg)) {
Log_debug(rap->log, "Out of space pushing metadata o_O");
} else if (pushWithdrawLinks(rap, msg)) {
Log_debug(rap->log, "Out of space pushing peer withdrawals");
} else if (pushPeers(rap, msg)) {
Log_debug(rap->log, "Out of space pushing peers");
} else if (pushLinkState(rap, msg)) {
Log_debug(rap->log, "Out of space pushing link state");
} else {
// Inch the tba up whenever there's a "small" message
if (Message_getLength(msg) < 500) { rap->timeBetweenAnns += 100; }
// Cap at 60 seconds, going over this requires changing when
// nodes are re-announced.
if (rap->timeBetweenAnns > 60000) { rap->timeBetweenAnns = 60000; }
break;
}
stateUpdate(rap, ReachabilityAnnouncer_State_MSGFULL);
// Cut the tba in half every time there's a MSGFULL
rap->timeBetweenAnns /= 2;
// minimum tba is 500ms
if (rap->timeBetweenAnns < 500) { rap->timeBetweenAnns = 500; }
} while (0);
Err_assert(Message_epush(msg, NULL, Announce_Header_SIZE));
struct Announce_Header* hdr = (struct Announce_Header*) Message_bytes(msg);
Bits_memset(hdr, 0, Announce_Header_SIZE);
Announce_Header_setVersion(hdr, Announce_Header_CURRENT_VERSION);
Announce_Header_setReset(hdr, rap->resetState);
Assert_true(Announce_Header_isReset(hdr) == rap->resetState);
Announce_Header_setTimestamp(hdr, snNow);
Bits_memcpy(hdr->pubSigningKey, rap->pubSigningKey, 32);
Bits_memcpy(hdr->snodeIp, rap->snode.ip6.bytes, 16);
Err_assert(Message_epop(msg, NULL, 64));
Sign_signMsg(rap->signingKeypair, msg, rap->rand);
Dict* dict = qp->msg = Dict_new(qp->alloc);
qp->cb = onReply;
struct Query* q = Allocator_calloc(qp->alloc, sizeof(struct Query), 1);
Identity_set(q);
q->rap = rap;
q->msg = msg;
Assert_true(AddressCalc_validAddress(rap->snode.ip6.bytes));
Bits_memcpy(&q->target, &rap->snode, Address_SIZE);
qp->userData = q;
qp->target = &q->target;
Dict_putStringCC(dict, "sq", "ann", qp->alloc);
String* annString = String_newBinary(Message_bytes(msg), Message_getLength(msg), qp->alloc);
Dict_putStringC(dict, "ann", annString, qp->alloc);
rap->onTheWire = q;
rap->msgOnWireSentTime = now;
}
static void onSnodeChange(struct SupernodeHunter* sh,
int64_t sendTime,
int64_t snodeRecvTime)
{
struct ReachabilityAnnouncer_pvt* rap =
Identity_check((struct ReachabilityAnnouncer_pvt*) sh->userData);
int64_t clockSkew = estimateClockSkew(sendTime, snodeRecvTime, ourTime(rap));
uint64_t clockSkewDiff = (clockSkew > rap->clockSkew)
? (clockSkew - rap->clockSkew)
: (rap->clockSkew - clockSkew);
// If the node is the same and the clock skew difference is less than 10 seconds,
// just change path and continue.
if (Bits_memcmp(rap->snode.key, sh->snodeAddr.key, 32)) {
if (Defined(Log_DEBUG)) {
uint8_t oldSnode[40];
AddrTools_printIp(oldSnode, rap->snode.ip6.bytes);
uint8_t newSnode[40];
AddrTools_printIp(newSnode, sh->snodeAddr.ip6.bytes);
Log_debug(rap->log, "Change Supernode [%s] -> [%s]", oldSnode, newSnode);
}
} else if (clockSkewDiff > 5000) {
Log_debug(rap->log,
"Change Supernode (no change but clock skew diff [%" PRIu64 "] > 5000ms)",
clockSkewDiff);
} else if (rap->snode.path == sh->snodeAddr.path) {
Log_debug(rap->log, "Change Supernode (not really, false call)");
return;
} else {
uint8_t oldPath[20];
uint8_t newPath[20];
AddrTools_printPath(oldPath, rap->snode.path);
AddrTools_printPath(newPath, sh->snodeAddr.path);
Log_debug(rap->log, "Change Supernode path [%s] -> [%s]", oldPath, newPath);
Bits_memcpy(&rap->snode, &sh->snodeAddr, Address_SIZE);
PeeringSeeder_setSnode(rap->ps, &rap->snode);
return;
}
Bits_memcpy(&rap->snode, &sh->snodeAddr, Address_SIZE);
PeeringSeeder_setSnode(rap->ps, &rap->snode);
rap->clockSkew = clockSkew;
stateReset(rap);
}
static struct Announce_ItemHeader* mkEncodingSchemeItem(
struct Allocator* alloc,
String* compressedScheme)
{
struct Allocator* tmpAlloc = Allocator_child(alloc);
Message_t* esMsg = Message_new(0, 256, tmpAlloc);
Assert_true(compressedScheme->len + 2 < 256);
Err_assert(Message_epush(esMsg, compressedScheme->bytes, compressedScheme->len));
Err_assert(Message_epush8(esMsg, Announce_Type_ENCODING_SCHEME));
Err_assert(Message_epush8(esMsg, compressedScheme->len + 2));
struct Announce_ItemHeader* item = Allocator_calloc(alloc, Message_getLength(esMsg), 1);
Bits_memcpy(item, Message_bytes(esMsg), Message_getLength(esMsg));
Allocator_free(tmpAlloc);
return item;
}
struct ReachabilityAnnouncer* ReachabilityAnnouncer_new(struct Allocator* allocator,
struct Log* log,
EventBase_t* base,
struct Random* rand,
struct MsgCore* msgCore,
struct SupernodeHunter* snh,
uint8_t* privateKey,
struct EncodingScheme* myScheme,
struct ReachabilityCollector* rc,
PeeringSeeder_t* ps)
{
struct Allocator* alloc = Allocator_child(allocator);
struct ReachabilityAnnouncer_pvt* rap =
Allocator_calloc(alloc, sizeof(struct ReachabilityAnnouncer_pvt), 1);
Identity_set(rap);
rap->alloc = alloc;
rap->log = log;
rap->base = base;
rap->msgCore = msgCore;
rap->announceCycle = Timeout_setInterval(onAnnounceCycle, rap, 250, base, alloc);
rap->rand = rand;
rap->snodeState = ArrayList_OfMessages_new(alloc);
rap->myScheme = myScheme;
rap->encodingSchemeStr = EncodingScheme_serialize(myScheme, alloc);
rap->rc = rc;
rap->ps = ps;
rap->mySchemeItem =
(struct Announce_ItemHeader*) mkEncodingSchemeItem(alloc, rap->encodingSchemeStr);
rap->snh = snh;
snh->onSnodeChange = onSnodeChange;
snh->userData = rap;
Sign_signingKeyPairFromCurve25519(rap->signingKeypair, privateKey);
Sign_publicKeyFromKeyPair(rap->pubSigningKey, rap->signingKeypair);
return &rap->pub;
}