diff --git a/include/constants.h b/include/constants.h index d25b923..beacf7d 100644 --- a/include/constants.h +++ b/include/constants.h @@ -82,7 +82,18 @@ static const size_t MAX_ORPHAN_BLOCKS = 512U; // following the tip, and is exempt from the reorg penalty (Horizen does the same via // IsInitialBlockDownload). Determined purely from local state, so an unverified peer cannot // trigger the exemption by claiming a large height. -static const uint64_t IBD_TIP_AGE_BLOCKS = 500ULL; +// +// This is also the ONLY way a non-mining node rejoins the network after ending up on a minority +// fork: the penalty is served by local chain growth, and a node that does not mine has no way to +// grow except by adopting the very branch the penalty is gating. It therefore has to be short +// enough that such a node recovers in minutes rather than half a day. +// +// 20 block times is ~30 minutes at a 90s target, far beyond normal Poisson block spacing (a gap +// that long has probability ~e^-20), so a node that is genuinely following the tip will not trip +// it. Note the exemption is all-or-nothing -- once in IBD a node accepts a reorg of any depth -- +// so lowering this further widens that hole; it is the number to revisit if deep reorgs ever get +// used against an idle node. +static const uint64_t IBD_TIP_AGE_BLOCKS = 20ULL; // Number of trailing blocks whose median timestamp is used for the age test above. Using a median // rather than the tip alone means a single miner cannot backdate one block to fake being in IBD. static const size_t MEDIAN_TIME_SPAN = 11U; diff --git a/include/nets/net_node.h b/include/nets/net_node.h index f45275f..0454f51 100644 --- a/include/nets/net_node.h +++ b/include/nets/net_node.h @@ -72,6 +72,29 @@ int Node_BroadcastTransaction(net_node_t* node, signed_transaction_t* tx, tcp_co // Helpers for outbound peer selection and block broadcast int Node_GetBestOutboundPeer(net_node_t* node, tcp_connection_t** outConn, uint64_t* outHeight); + +/** + * Delivery receipts for windowed sync. + * + * A FETCH_BLOCK reply is handled on the peer's io thread and may legitimately never reach the + * chain: a block belonging to a competing branch is filed in the orphan pool instead. A sync loop + * that infers arrival from the chain growing therefore cannot tell "arrived but forked" from "lost + * in transit", so it re-requests until it times out. Against a peer on a fork that costs one full + * retry-and-timeout cycle for EVERY block, which is why syncing to a forked peer used to crawl. + * + * DUPLICATE is what makes a backwards fork walk terminate: it means we already hold exactly that + * block, so the two chains agree at that height and there is no reason to keep descending. +**/ +typedef enum { + NODE_DELIVERY_APPENDED = 0, // joined our chain + NODE_DELIVERY_DUPLICATE = 1, // we already held this exact block -- common ground + NODE_DELIVERY_ORPHANED = 2, // belongs to a competing branch; now in the orphan pool + NODE_DELIVERY_REJECTED = 3 // failed validation +} node_delivery_status_t; + +void Node_NoteBlockDelivered(uint64_t height, node_delivery_status_t status); +bool Node_TakeBlockDelivery(uint64_t height, node_delivery_status_t* outStatus); +void Node_ResetBlockDeliveries(void); void Node_BroadcastChainRange(net_node_t* node, size_t startHeightInclusive, tcp_connection_t* sourceConn); // Callback logic diff --git a/src/block/chain.c b/src/block/chain.c index 2edae25..f62776a 100644 --- a/src/block/chain.c +++ b/src/block/chain.c @@ -842,7 +842,9 @@ bool Chain_ReplaceBranch(blockchain_t* chain, do { const size_t tipCount = DynArr_size(chain->blocks); if (forkHeight > tipCount) { - break; // fork point is beyond our chain; nothing to replace + printf("Chain_ReplaceBranch: fork point %zu is beyond our tip %zu; nothing to replace\n", + forkHeight, tipCount); + break; } if (!Chain_BranchIsLinkedLocked(chain, forkHeight, newBlocks, count)) { @@ -883,10 +885,17 @@ bool Chain_ReplaceBranch(blockchain_t* chain, uint256_t candidateWork; if (!Chain_ComputeWorkRange(chain, forkHeight, tipCount, &incumbentWork) || !Chain_ComputeBranchWork(newBlocks, count, &candidateWork)) { + printf("Chain_ReplaceBranch: could not compute work for the branch at height %zu\n", forkHeight); break; } if (uint256_cmp(&candidateWork, &incumbentWork) <= 0) { - break; // not heavier; keep what we have + // Very often this just means the branch is still arriving -- a partial branch is + // genuinely lighter than what it would replace. Report the block counts so that case + // is distinguishable from a peer that really is on a weaker chain. + printf("Chain_ReplaceBranch: candidate at height %zu is not heavier " + "(%zu candidate block(s) vs %zu incumbent); keeping our chain\n", + forkHeight, count, tipCount - forkHeight); + break; } // Snapshot what we are about to discard so a failed apply can be undone. The in-memory @@ -896,6 +905,8 @@ bool Chain_ReplaceBranch(blockchain_t* chain, if (snapshotCount > 0) { snapshot = (block_t**)calloc(snapshotCount, sizeof(block_t*)); if (!snapshot) { + printf("Chain_ReplaceBranch: out of memory snapshotting %zu block(s); chain unchanged\n", + snapshotCount); break; } @@ -915,6 +926,10 @@ bool Chain_ReplaceBranch(blockchain_t* chain, } } if (!snapshotOk) { + // Refusing here is the point: without a complete snapshot a failed apply could not + // be undone, so we would rather not start than risk a half-replaced chain. + printf("Chain_ReplaceBranch: could not snapshot the blocks being replaced at height %zu; " + "refusing the reorg rather than risk an unrecoverable apply\n", forkHeight); break; } } @@ -923,6 +938,7 @@ bool Chain_ReplaceBranch(blockchain_t* chain, // rollback -- the caller keeps ownership of what it passed in, whatever happens here. candidate = (block_t**)calloc(count, sizeof(block_t*)); if (!candidate) { + printf("Chain_ReplaceBranch: out of memory copying %zu candidate block(s); chain unchanged\n", count); break; } bool copiedAll = true; @@ -934,6 +950,8 @@ bool Chain_ReplaceBranch(blockchain_t* chain, } } if (!copiedAll) { + printf("Chain_ReplaceBranch: could not copy the candidate branch at height %zu; chain unchanged\n", + forkHeight); break; } diff --git a/src/main.c b/src/main.c index 4bb801d..d6d096a 100644 --- a/src/main.c +++ b/src/main.c @@ -361,26 +361,104 @@ static bool Block_GetCoinbaseAndFeeTotals(const block_t* block, uint64_t* outCoi * common ancestor by prevHash linkage and Chain_ReplaceBranch decides whether to adopt. FETCH_BLOCK * already answers from the peer's own chain, so finding a fork needs no new packet type. **/ -static void RequestForkWindow(net_node_t* node, tcp_connection_t* peerConn, uint64_t topHeight) { +/** + * Walk backwards from `topHeight` pulling the peer's blocks until we reach common ground, so the + * orphan pool holds a branch that links to a block we already have. + * + * Two things this does that a flat "ask for REORG_FETCH_DEPTH blocks and sleep" did not: + * + * - It STOPS at the fork point. A block the peer returns that we already hold means the chains + * agree there and everything below is shared, so there is nothing left to ask for. The old + * version always requested the full depth; on a shallow fork the overwhelming majority came + * back as duplicates, were discarded without even entering the pool, and cost the peer a full + * block send each. + * + * - It waits on delivery receipts instead of a fixed sleep. The flat wait was a race against the + * peer's serving rate -- at ~10 blocks/s a 128-block window cannot land in 5s -- so the attach + * that followed ran against a half-filled pool and reported a failure that was not real. + * + * Returns true if a shared block was found (so the pool should have a linkable branch). +**/ +static bool RequestForkWindow(net_node_t* node, tcp_connection_t* peerConn, uint64_t topHeight) { if (!node || !peerConn) { - return; + return false; } - const uint64_t from = (topHeight > REORG_FETCH_DEPTH) ? (topHeight - REORG_FETCH_DEPTH) : 0ULL; - printf("Requesting peer blocks %" PRIu64 "..%" PRIu64 " to locate the fork point\n", from, topHeight); + const uint64_t floorHeight = (topHeight > REORG_FETCH_DEPTH) ? (topHeight - REORG_FETCH_DEPTH) : 0ULL; + printf("Walking back from %" PRIu64 " (floor %" PRIu64 ") to locate the fork point\n", + topHeight, floorHeight); - for (uint64_t hh = topHeight + 1; hh-- > from; ) { - uint64_t req = hh; - if (Node_SendPacket(node, peerConn, PACKET_TYPE_FETCH_BLOCK, &req, sizeof(req)) != 0) { + uint64_t batch[64]; + node_delivery_status_t status[64]; + bool got[64]; + int batchMax = MAX_PARALLEL_FETCHES; + if (batchMax > (int)(sizeof(batch) / sizeof(batch[0]))) { + batchMax = (int)(sizeof(batch) / sizeof(batch[0])); + } + + uint64_t totalRequested = 0; + uint64_t next = topHeight; + bool exhausted = false; + + while (!exhausted) { + int batchCount = 0; + while (batchCount < batchMax) { + uint64_t req = next; + if (Node_SendPacket(node, peerConn, PACKET_TYPE_FETCH_BLOCK, &req, sizeof(req)) != 0) { + exhausted = true; + break; + } + batch[batchCount] = req; + got[batchCount] = false; + status[batchCount] = NODE_DELIVERY_REJECTED; + batchCount++; + totalRequested++; + + if (req == floorHeight) { + exhausted = true; + break; + } + next = req - 1ULL; + } + + if (batchCount == 0) { break; } - if (hh == 0) { - break; + + // Wait for this batch specifically, rather than guessing how long the peer needs. + const uint64_t deadline = get_current_time_ms() + SYNC_REQUEST_TIMEOUT_MS; + int outstanding = batchCount; + while (outstanding > 0 && get_current_time_ms() < deadline) { + for (int i = 0; i < batchCount; ++i) { + if (!got[i] && Node_TakeBlockDelivery(batch[i], &status[i])) { + got[i] = true; + outstanding--; + } + } + if (outstanding > 0) { + sleep_for_milliseconds(20); + } + } + + // The batch descends, so the first height we already hold is the boundary between the + // shared prefix and the competing branch. + for (int i = 0; i < batchCount; ++i) { + if (got[i] && (status[i] == NODE_DELIVERY_DUPLICATE || status[i] == NODE_DELIVERY_APPENDED)) { + printf("fork walk: chains agree at height %" PRIu64 " after %" PRIu64 " request(s)\n", + batch[i], totalRequested); + return true; + } + } + + if (outstanding > 0) { + printf("fork walk: %d of %d block(s) unanswered; stopping the descent\n", outstanding, batchCount); + return false; } } - // Give the asynchronous replies time to land in the pool. - sleep_for_milliseconds(SYNC_REQUEST_TIMEOUT_MS); + printf("fork walk: no shared block within %" PRIu64 " request(s) of height %" PRIu64 "\n", + totalRequested, topHeight); + return false; } static bool MineAndAppendBlock(blockchain_t* chain, @@ -1095,6 +1173,11 @@ int main(int argc, char* argv[]) { // Continue syncing in a loop until we've caught up to the peer or no progress is made. bool madeProgressOverall = false; int forkProbes = 0; + + // Drop receipts left over from an earlier sync, so a stale one cannot be mistaken for + // an answer to a request this run has not sent yet. + Node_ResetBlockDeliveries(); + while (true) { uint64_t localHeight = (uint64_t)Chain_Size(chain); @@ -1243,6 +1326,62 @@ int main(int argc, char* argv[]) { break; // restart loop to re-evaluate } + // The peer answered, but the block never joined our chain -- it is on a + // competing branch and now sits in the orphan pool. Retrying cannot change + // that, and the old code could not tell this case from a dropped packet: it + // burned MAX_SYNC_RETRIES plus a timeout on EVERY block, then slid the window + // forward and did it again, which is what made syncing to a forked peer crawl + // and made the peer re-serve the whole chain several times over. + node_delivery_status_t deliveryStatus = NODE_DELIVERY_REJECTED; + if (Node_TakeBlockDelivery(h, &deliveryStatus) && deliveryStatus != NODE_DELIVERY_APPENDED) { + if (forkProbes >= MAX_FORK_PROBE_ROUNDS) { + printf("block %" PRIu64 " is on a branch we cannot join after %d probe(s); " + "giving up on this peer\n", h, forkProbes); + inFlight = 0; + nextReq = end; // stop refilling the window; this peer is unreachable by extension + break; + } + + forkProbes++; + printf("block %" PRIu64 " arrived but does not extend our chain; " + "probing for the fork point (round %d/%d)\n", + h, forkProbes, MAX_FORK_PROBE_ROUNDS); + + // Pull a window BELOW the divergence so the pool can walk prevHash back to + // the common ancestor. Without this the fork point is never requested at + // all, because the window only ever moves forward from our own tip. + const bool foundCommon = RequestForkWindow(node, peerConn, h); + + size_t reattached = foundCommon ? OrphanPool_AttemptAttach(chain) : 0; + if (reattached > 0) { + printf("Reorg attached %zu block(s) from the peer's branch\n", reattached); + forkProbes = 0; // real progress; allow probing again if it forks further on + } else if (!foundCommon) { + printf("No shared block found with this peer; its branch cannot be linked to ours\n"); + } else { + // Deliberately not phrased as a failure: the branch stays pooled and the + // 1Hz maintenance thread retries it, which is usually what completes a + // reorg whose blocks were still arriving when this pass ran. + printf("Reorg not completed on this pass; branch stays pooled for retry\n"); + } + + // Restart the window against whatever our tip is now. + nextReq = Chain_Size(chain); + inFlight = 0; + if (Chain_Size(chain) > 0) { + block_t* tip = NULL; + if (Chain_GetBlockCopy(chain, Chain_Size(chain) - 1, &tip) && tip) { + Block_CalculateHash(tip, expectedPrevHash); + Block_Destroy(tip); + } + } else { + memset(expectedPrevHash, 0, sizeof(expectedPrevHash)); + } + + progressed = true; + break; + } + uint64_t elapsed = (now > sentAtMs[i]) ? (now - sentAtMs[i]) : 0ULL; if (elapsed > SYNC_REQUEST_TIMEOUT_MS) { if (retryCount[i] < MAX_SYNC_RETRIES) { @@ -1304,14 +1443,18 @@ int main(int argc, char* argv[]) { forkProbes++; printf("No progress but peer is ahead (%" PRIu64 " > %" PRIu64 "); probing for a fork point\n", peerHeight, newLocal); - RequestForkWindow(node, peerConn, newLocal); + const bool foundCommon = RequestForkWindow(node, peerConn, newLocal); - size_t attached = OrphanPool_AttemptAttach(chain); + size_t attached = foundCommon ? OrphanPool_AttemptAttach(chain) : 0; if (attached > 0) { printf("Fork probe adopted %zu block(s) from the peer's branch\n", attached); continue; } - printf("Fork probe found nothing adoptable (lighter branch, or still serving its reorg penalty)\n"); + if (!foundCommon) { + printf("Fork probe found no shared block with this peer\n"); + } else { + printf("Fork probe did not complete a reorg; branch stays pooled for retry\n"); + } } break; } diff --git a/src/nets/net_node.c b/src/nets/net_node.c index bea4982..28b00a8 100644 --- a/src/nets/net_node.c +++ b/src/nets/net_node.c @@ -349,6 +349,67 @@ typedef enum { NODE_BLOCK_DUPLICATE = 3 // already on our chain; not a fault, do not log it as a rejection } node_block_accept_result_t; +// Delivery receipts for windowed sync -- see the contract in net_node.h. Written from peer io +// threads, drained by the REPL thread running `sync`, so it needs its own lock; it never calls back +// into chain.c or takes any other lock, so it cannot participate in a cycle. +#define NODE_DELIVERY_SLOTS 512 +typedef struct { + uint64_t height; + node_delivery_status_t status; + bool valid; +} node_delivery_t; + +static node_delivery_t g_deliveries[NODE_DELIVERY_SLOTS]; +static size_t g_deliveryNext = 0; +static pthread_mutex_t g_deliveryLock = PTHREAD_MUTEX_INITIALIZER; + +void Node_NoteBlockDelivered(uint64_t height, node_delivery_status_t status) { + pthread_mutex_lock(&g_deliveryLock); + + // Refresh an existing receipt rather than adding a second one for the same height: a retried + // request would otherwise leave a stale receipt that the next window could consume by mistake. + for (size_t i = 0; i < NODE_DELIVERY_SLOTS; ++i) { + if (g_deliveries[i].valid && g_deliveries[i].height == height) { + g_deliveries[i].status = status; + pthread_mutex_unlock(&g_deliveryLock); + return; + } + } + + g_deliveries[g_deliveryNext].height = height; + g_deliveries[g_deliveryNext].status = status; + g_deliveries[g_deliveryNext].valid = true; + g_deliveryNext = (g_deliveryNext + 1u) % NODE_DELIVERY_SLOTS; + + pthread_mutex_unlock(&g_deliveryLock); +} + +bool Node_TakeBlockDelivery(uint64_t height, node_delivery_status_t* outStatus) { + bool found = false; + + pthread_mutex_lock(&g_deliveryLock); + for (size_t i = 0; i < NODE_DELIVERY_SLOTS; ++i) { + if (g_deliveries[i].valid && g_deliveries[i].height == height) { + if (outStatus) { + *outStatus = g_deliveries[i].status; + } + g_deliveries[i].valid = false; // consumed + found = true; + break; + } + } + pthread_mutex_unlock(&g_deliveryLock); + + return found; +} + +void Node_ResetBlockDeliveries(void) { + pthread_mutex_lock(&g_deliveryLock); + memset(g_deliveries, 0, sizeof(g_deliveries)); + g_deliveryNext = 0; + pthread_mutex_unlock(&g_deliveryLock); +} + // Reclaims outbound slots whose peer has disconnected. Mirrors the inbound self-reclaim in // TcpServer_clientthreadprocess: detach dead connections from their slots under outboundLock, then // join their io threads and destroy/free them outside the lock. Pinned connections (a raw pointer @@ -1362,6 +1423,19 @@ void Node_Client_OnData(tcp_connection_t* client) { uint64_t blockHeight = 0; memcpy(&blockHeight, payload, sizeof(blockHeight)); node_block_accept_result_t result = Node_ParseAndAcceptBlock(payload, payloadLen, true); + + // Receipt for the windowed sync. BLOCK_DATA is only ever sent in reply to a + // FETCH_BLOCK, so recording it here (and not for BROADCAST_BLOCK) tells the sync + // loop the peer answered, whether or not the block could join our chain. + node_delivery_status_t deliveryStatus = NODE_DELIVERY_REJECTED; + switch (result) { + case NODE_BLOCK_ACCEPTED: deliveryStatus = NODE_DELIVERY_APPENDED; break; + case NODE_BLOCK_DUPLICATE: deliveryStatus = NODE_DELIVERY_DUPLICATE; break; + case NODE_BLOCK_ORPHAN_QUEUED: deliveryStatus = NODE_DELIVERY_ORPHANED; break; + default: deliveryStatus = NODE_DELIVERY_REJECTED; break; + } + Node_NoteBlockDelivered(blockHeight, deliveryStatus); + if (result == NODE_BLOCK_ACCEPTED) { printf("Accepted BLOCK_DATA from node %u\n", client ? client->connectionId : 0U); net_node_t* node = Node_FromConnection(client);