/* 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; }