diff --git a/include/constants.h b/include/constants.h index 6358779..2960bf1 100644 --- a/include/constants.h +++ b/include/constants.h @@ -121,9 +121,12 @@ static const uint64_t TAIL_EMISSION = 750000000000ULL; // 0.75 coins per block f // No max supply. Instead of halving, it'll follow a more gradual, Monero-like emission curve. // Phase 3: update once per effective epoch and keep a fixed per-block reward for that epoch. -static inline uint64_t GetInflationRateReward(uint256_t currentSupply, blockchain_t* chain) { - if (!chain || !chain->blocks) { return 0x00; } // Invalid - size_t height = Chain_Size(chain); +// +// The *AtHeight variants take the height directly and never call Chain_Size/Chain_GetBlockCopy, so +// they are safe to call from inside a chainLock critical section. chainLock is a non-recursive +// pthread_rwlock_t: taking it for reading while this thread already holds it for writing deadlocks +// as soon as another thread is queued for the write lock. +static inline uint64_t GetInflationRateRewardAtHeight(uint256_t currentSupply, uint64_t height) { const uint64_t effectiveEpochLength = (EPOCH_LENGTH / EMISSION_ACCELERATION_FACTOR) > 0 ? (EPOCH_LENGTH / EMISSION_ACCELERATION_FACTOR) @@ -161,27 +164,29 @@ static inline uint64_t GetInflationRateReward(uint256_t currentSupply, blockchai return (currentReward > TAIL_EMISSION) ? currentReward : TAIL_EMISSION; } -static inline uint64_t CalculateBlockReward(uint256_t currentSupply, blockchain_t* chain) { +static inline uint64_t GetInflationRateReward(uint256_t currentSupply, blockchain_t* chain) { if (!chain || !chain->blocks) { return 0x00; } // Invalid + return GetInflationRateRewardAtHeight(currentSupply, (uint64_t)Chain_Size(chain)); +} +static inline uint64_t CalculateBlockRewardAtHeight(uint256_t currentSupply, uint64_t height) { const uint64_t effectivePhase1Blocks = (PHASE1_TARGET_BLOCKS / EMISSION_ACCELERATION_FACTOR) > 0 ? (PHASE1_TARGET_BLOCKS / EMISSION_ACCELERATION_FACTOR) : 1; - const uint64_t height = (uint64_t)Chain_Size(chain); // After the phase-one target horizon, only floor/inflation schedule applies. if (height >= effectivePhase1Blocks) { - return GetInflationRateReward(currentSupply, chain); + return GetInflationRateRewardAtHeight(currentSupply, height); } - if (currentSupply.limbs[1] > 0 || - currentSupply.limbs[2] > 0 || - currentSupply.limbs[3] > 0 || + if (currentSupply.limbs[1] > 0 || + currentSupply.limbs[2] > 0 || + currentSupply.limbs[3] > 0 || currentSupply.limbs[0] >= M_CAP) { // Post-Monero phase with unlimited supply: floor/inflation schedule only. - return GetInflationRateReward(currentSupply, chain); + return GetInflationRateRewardAtHeight(currentSupply, height); } const uint64_t generated = currentSupply.limbs[0]; @@ -213,7 +218,12 @@ static inline uint64_t CalculateBlockReward(uint256_t currentSupply, blockchain_ } // Phase 2 + 3: floor and epoch inflation updates. - return GetInflationRateReward(currentSupply, chain); + return GetInflationRateRewardAtHeight(currentSupply, height); +} + +static inline uint64_t CalculateBlockReward(uint256_t currentSupply, blockchain_t* chain) { + if (!chain || !chain->blocks) { return 0x00; } // Invalid + return CalculateBlockRewardAtHeight(currentSupply, (uint64_t)Chain_Size(chain)); } // Hashing DAG diff --git a/src/block/chain.c b/src/block/chain.c index a9b899f..4c812f5 100644 --- a/src/block/chain.c +++ b/src/block/chain.c @@ -176,7 +176,8 @@ bool Chain_RecomputeRuntimeState(blockchain_t* chain) { } currentSupply = rebuiltSupply; - currentReward = CalculateBlockReward(currentSupply, chain); + // *AtHeight: never take chainLock from here, callers may already hold it. + currentReward = CalculateBlockRewardAtHeight(currentSupply, (uint64_t)chain->size); return true; } @@ -420,7 +421,10 @@ static bool Chain_AddBlockLocked(blockchain_t* chain, block_t* block) { // forever. It also has to happen per block so that applying a whole branch works. if (ok) { (void)uint256_add_u64(¤tSupply, expectedCoinbaseAmount); - currentReward = CalculateBlockReward(currentSupply, chain); + // Must be the *AtHeight variant: we hold chainLock for writing here, and + // CalculateBlockReward would take it for reading via Chain_Size. chainLock is not + // recursive, so that self-deadlocks as soon as another thread queues for the write lock. + currentReward = CalculateBlockRewardAtHeight(currentSupply, (uint64_t)chain->size); } // ok remains true if no failures } while (0); @@ -639,7 +643,8 @@ static bool Chain_RollbackToHeightLocked(blockchain_t* chain, size_t height) { } currentSupply = rebuiltSupply; - currentReward = CalculateBlockReward(currentSupply, chain); + // *AtHeight: chainLock is held for writing here (see Chain_RollbackToHeight). + currentReward = CalculateBlockRewardAtHeight(currentSupply, (uint64_t)chain->size); return true; } diff --git a/src/nets/net_node.c b/src/nets/net_node.c index 3efd8e5..04581b6 100644 --- a/src/nets/net_node.c +++ b/src/nets/net_node.c @@ -345,7 +345,8 @@ static void Node_OnPingTimeoutThunk(udp_node_t* udp, const struct sockaddr_stora typedef enum { NODE_BLOCK_REJECTED = 0, NODE_BLOCK_ORPHAN_QUEUED = 1, - NODE_BLOCK_ACCEPTED = 2 + NODE_BLOCK_ACCEPTED = 2, + NODE_BLOCK_DUPLICATE = 3 // already on our chain; not a fault, do not log it as a rejection } node_block_accept_result_t; // Reclaims outbound slots whose peer has disconnected. Mirrors the inbound self-reclaim in @@ -511,7 +512,7 @@ static node_block_accept_result_t Node_ParseAndAcceptBlock(const unsigned char* // Exactly the block we already have. DynArr_destroy(blk->transactions); free(blk); - return NODE_BLOCK_REJECTED; + return NODE_BLOCK_DUPLICATE; } } @@ -1105,6 +1106,8 @@ void Node_Server_OnData(tcp_connection_t* client) { } } else if (result == NODE_BLOCK_ORPHAN_QUEUED) { printf("Queued orphan BROADCAST_BLOCK from node %u\n", client ? client->connectionId : 0U); + } else if (result == NODE_BLOCK_DUPLICATE) { + // Already on our chain (a peer relayed it to us twice); not an error. } else { printf("Rejected BROADCAST_BLOCK from node %u\n", client ? client->connectionId : 0U); } @@ -1344,6 +1347,8 @@ void Node_Client_OnData(tcp_connection_t* client) { } } else if (result == NODE_BLOCK_ORPHAN_QUEUED) { printf("Queued orphan BLOCK_DATA from node %u\n", client ? client->connectionId : 0U); + } else if (result == NODE_BLOCK_DUPLICATE) { + // Already on our chain (a peer relayed it to us twice); not an error. } else { printf("Rejected BLOCK_DATA from node %u\n", client ? client->connectionId : 0U); } @@ -1375,6 +1380,8 @@ void Node_Client_OnData(tcp_connection_t* client) { } } else if (result == NODE_BLOCK_ORPHAN_QUEUED) { printf("Queued orphan BROADCAST_BLOCK from node %u\n", client ? client->connectionId : 0U); + } else if (result == NODE_BLOCK_DUPLICATE) { + // Already on our chain (a peer relayed it to us twice); not an error. } else { printf("Rejected BROADCAST_BLOCK from node %u\n", client ? client->connectionId : 0U); } @@ -1536,38 +1543,81 @@ void Node_BroadcastChainRange(net_node_t* node, size_t startHeightInclusive, tcp memcpy(payload + off, tx, sizeof(signed_transaction_t)); off += sizeof(signed_transaction_t); } - size_t delivered = 0; + // Collect one connection per distinct peer, then send with no lock held. + // + // A peer we both dialled and were dialled by occupies two connections (one outbound, one + // inbound). Sending on both delivers every block twice, and the receiver logs the second + // copy as a rejection. Peers are identified by peerNodeId rather than by endpoint, because + // a multi-homed host reaches us from several addresses and an inbound connection carries an + // ephemeral port while the outbound one carries the listen port. + // + // Sends happen outside outboundLock/clientsMutex on purpose: Node_SendPacket writes to a + // socket and can block when the peer is slow to read, and holding the server's clientsMutex + // across that stalls the accept path and every other user of it. + tcp_connection_t* targets[MAX_CONS * 2]; + uint64_t targetNodeIds[MAX_CONS * 2]; + size_t targetCount = 0; + + uint64_t sourceNodeId = sourceConn ? sourceConn->peerNodeId : 0ULL; + + // Skip a connection if it is the source, belongs to the source's node, or duplicates a peer + // we have already queued. + #define NODE_RELAY_SHOULD_SKIP(conn) ( \ + (conn) == sourceConn || \ + ((sourceNodeId != 0ULL) && ((conn)->peerNodeId == sourceNodeId)) || \ + (sourceConn && (sourceNodeId == 0ULL) && TcpConnection_PeerAddrEqual((conn), sourceConn))) - // Snapshot outbound clients and send pthread_mutex_lock(&node->outboundLock); - for (size_t i = 0; i < MAX_CONS; ++i) { + for (size_t i = 0; i < MAX_CONS && targetCount < (MAX_CONS * 2); ++i) { tcp_connection_t* conn = node->outboundClients[i].connection; - if (!conn) continue; - if (conn == sourceConn) continue; - if (sourceConn && TcpConnection_PeerAddrEqual(conn, sourceConn)) continue; - if (Node_SendPacket(node, conn, PACKET_TYPE_BROADCAST_BLOCK, payload, off) == 0) { - delivered++; + if (!conn || TcpConnection_IsDisconnectNotified(conn)) continue; + if (NODE_RELAY_SHOULD_SKIP(conn)) continue; + + bool duplicate = false; + for (size_t t = 0; t < targetCount; ++t) { + if (conn->peerNodeId != 0ULL && targetNodeIds[t] == conn->peerNodeId) { duplicate = true; break; } } + if (duplicate) continue; + + TcpConnection_Pin(conn); + targetNodeIds[targetCount] = conn->peerNodeId; + targets[targetCount++] = conn; } pthread_mutex_unlock(&node->outboundLock); - // Relay to inbound peers as well. Broadcasting only to outbound connections meant that in a - // two-node setup the node that was dialled never pushed anything back, and the dialer only - // ever learned about new blocks through a manual `sync`. + // Inbound peers too. Broadcasting only to outbound connections meant that in a two-node + // setup the node that was dialled never pushed anything back, and the dialer only ever + // learned about new blocks through a manual `sync`. if (node->server) { pthread_mutex_lock(&node->server->clientsMutex); - for (size_t i = 0; i < node->server->maxClients; ++i) { + for (size_t i = 0; i < node->server->maxClients && targetCount < (MAX_CONS * 2); ++i) { tcp_connection_t* conn = node->server->clientsArrPtr ? node->server->clientsArrPtr[i] : NULL; if (!conn || TcpConnection_IsDisconnectNotified(conn)) continue; - if (conn == sourceConn) continue; - if (sourceConn && TcpConnection_PeerAddrEqual(conn, sourceConn)) continue; - if (Node_SendPacket(node, conn, PACKET_TYPE_BROADCAST_BLOCK, payload, off) == 0) { - delivered++; + if (NODE_RELAY_SHOULD_SKIP(conn)) continue; + + bool duplicate = false; + for (size_t t = 0; t < targetCount; ++t) { + if (conn->peerNodeId != 0ULL && targetNodeIds[t] == conn->peerNodeId) { duplicate = true; break; } } + if (duplicate) continue; + + TcpConnection_Pin(conn); + targetNodeIds[targetCount] = conn->peerNodeId; + targets[targetCount++] = conn; } pthread_mutex_unlock(&node->server->clientsMutex); } + #undef NODE_RELAY_SHOULD_SKIP + + size_t delivered = 0; + for (size_t t = 0; t < targetCount; ++t) { + if (Node_SendPacket(node, targets[t], PACKET_TYPE_BROADCAST_BLOCK, payload, off) == 0) { + delivered++; + } + TcpConnection_Unpin(targets[t]); + } + if (delivered > 0) { pthread_mutex_lock(&node->seenLock); DynSet_Insert(node->seenBlocks, hash);