Fix self-deadlock on chainLock and fix double-delivery of accepted blocks (+ DUPLICATE enum entry)

2026-07-28 17:22:48 +02:00
parent 1ff2890c0f
commit 42b325d57a
3 changed files with 97 additions and 32 deletions
@@ -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,18 +164,20 @@ 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 ||
@@ -181,7 +186,7 @@ static inline uint64_t CalculateBlockReward(uint256_t currentSupply, blockchain_
        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
@@ -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(&currentSupply, 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;
 }
@@ -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);