sami7777/triangles_v5
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: compact blocks, column families, fork detector, cross-network discovery, SAM v3, configurable peers

Browse Source 
BIP152 Compact Blocks (main.cpp, net.cpp, protocol.h):
- SipHash-2-4 short IDs (48-bit) for transaction identification
- Compact block relay with mempool reconstruction
- Merkle root verification before acceptance
- Graceful fallback to full block on any mismatch
- Collision detection for ambiguous short IDs

RocksDB Column Families (txdb-rocksdb.cpp/h):
- 5 CFs: default, blockindex, txindex, utxo, addrindex
- Per-CF tuning: UTXO optimized for point lookups, addrindex for scans
- Backward-compatible: falls back to default CF for pre-migration data
- Prefix-based routing in ReadRaw/WriteRaw/EraseRaw/ExistsRaw

Fork Detector (main.cpp, net.cpp, net.h):
- Background thread checks local tip vs peer median every 60s post-IBD
- Alerts on divergence > forkthreshold (default 5 blocks)
- Optional auto-rebuild trigger on severe divergence

Cross-Network Tor↔I2P Discovery (net.cpp, init.cpp):
- I2P seed addresses loaded into addrman alongside onion seeds
- Address relay bridges .onion and .b32.i2p between networks
- IsI2PAddr/IsOnionAddr helpers for network-type detection

Configurable Outbound Connections (net.cpp, init.cpp):
- -maxoutboundconnections flag (range 4-32, default 8)

Mempool Fee-Priority Boost (miner.cpp):
- 2x fee weight in PoS block assembly for higher staking rewards

SAM v3 Direct Streaming (i2p/i2p_embedded.cpp/h):
- CI2PSamSocket class with full SAM v3 protocol
- SESSION CREATE + STREAM CONNECT handshake
- Factory method on CI2PEmbedded for native I2P connections
- SAM bridge readiness check in bootstrap loop
This commit is contained in:
Krystie
2026-06-27 19:19:30 -07:00
parent b623396186
commit fb07d50235
10 changed files with 1165 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/i2p/i2p_embedded.cpp
+355 -24
View File
@@ -33,6 +33,256 @@
namespace fs = std::filesystem;
// ===========================================================================
// CI2PSamSocket — SAM v3 direct streaming implementation
// ===========================================================================
//
// Protocol reference: https://geti2p.net/en/docs/api/samv3
//
// The SAM bridge is a simple line-oriented text protocol over TCP. After
// HELLO + SESSION CREATE + STREAM CONNECT succeed, the socket becomes a
// raw bidirectional byte stream to the I2P destination — no further SAM
// framing is needed and there is zero SOCKS overhead.
static std::atomic<unsigned int> g_samSessionSeq{0};
CI2PSamSocket::CI2PSamSocket()
: rawSocket(I2P_INVALID_SOCKET)
{
}
CI2PSamSocket::~CI2PSamSocket()
{
CloseSocket();
}
void CI2PSamSocket::CloseSocket()
{
if (rawSocket != I2P_INVALID_SOCKET) {
#ifdef WIN32
closesocket(rawSocket);
#else
close(rawSocket);
#endif
rawSocket = I2P_INVALID_SOCKET;
}
}
I2pSocket_t CI2PSamSocket::GetRawSocket()
{
I2pSocket_t fd = rawSocket;
rawSocket = I2P_INVALID_SOCKET; // transfer ownership
return fd;
}
bool CI2PSamSocket::SamConnect(const std::string& host, int port)
{
CloseSocket();
#ifdef WIN32
rawSocket = (I2pSocket_t)::socket(AF_INET, SOCK_STREAM, 0);
if (rawSocket == INVALID_SOCKET) {
#else
rawSocket = ::socket(AF_INET, SOCK_STREAM, 0);
if (rawSocket < 0) {
#endif
lastError = "SAM: failed to create socket";
return false;
}
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); // SAM is always local
addr.sin_port = htons((uint16_t)port);
if (::connect(rawSocket, (struct sockaddr*)&addr, sizeof(addr)) != 0) {
lastError = "SAM: cannot connect to bridge at 127.0.0.1:" + std::to_string(port);
CloseSocket();
return false;
}
return true;
}
bool CI2PSamSocket::SendLine(const std::string& line)
{
std::string msg = line + "\n";
const char* data = msg.data();
size_t remaining = msg.size();
while (remaining > 0) {
#ifdef WIN32
int n = ::send(rawSocket, data, (int)remaining, 0);
#else
ssize_t n = ::send(rawSocket, data, remaining, MSG_NOSIGNAL);
#endif
if (n <= 0) {
lastError = "SAM: send failed";
return false;
}
data += n;
remaining -= (size_t)n;
}
return true;
}
bool CI2PSamSocket::ReadLine(std::string& lineOut)
{
// Look for a complete line (terminated by \n) in recvBuffer first.
for (;;) {
size_t nl = recvBuffer.find('\n');
if (nl != std::string::npos) {
lineOut = recvBuffer.substr(0, nl);
// Strip trailing \r (SAM bridge always uses \n, but be tolerant)
if (!lineOut.empty() && lineOut.back() == '\r')
lineOut.pop_back();
recvBuffer.erase(0, nl + 1);
return true;
}
char buf[4096];
#ifdef WIN32
int n = ::recv(rawSocket, buf, sizeof(buf), 0);
#else
ssize_t n = ::recv(rawSocket, buf, sizeof(buf), 0);
#endif
if (n <= 0) {
lastError = "SAM: connection closed while waiting for reply";
return false;
}
recvBuffer.append(buf, (size_t)n);
}
}
std::string CI2PSamSocket::ParseValue(const std::string& line, const std::string& key)
{
// Find KEY=VALUE token within a space-separated SAM response line.
std::string needle = key + "=";
size_t pos = line.find(needle);
if (pos == std::string::npos)
return {};
pos += needle.size();
size_t end = line.find(' ', pos);
if (end == std::string::npos)
return line.substr(pos);
return line.substr(pos, end - pos);
}
bool CI2PSamSocket::Connect(const std::string& dest_b32, int port,
const std::string& samHost, int samPort)
{
CloseSocket();
lastError.clear();
recvBuffer.clear();
if (dest_b32.empty()) {
lastError = "SAM: empty destination";
return false;
}
// Generate a unique session ID for this connection.
unsigned int seq = ++g_samSessionSeq;
sessionId = "triangles-" + std::to_string(seq) + "-" +
std::to_string((unsigned long)std::time(nullptr));
// ----------------------------------------------------------------
// Step 0: TCP connect to the SAM bridge
// ----------------------------------------------------------------
if (!SamConnect(samHost, samPort)) {
// lastError already set by SamConnect
return false;
}
// ----------------------------------------------------------------
// Step 1: HELLO handshake
// C → S: HELLO VERSION MIN=3.1 MAX=3.1
// S → C: HELLO REPLY RESULT=OK VERSION=3.1
// ----------------------------------------------------------------
if (!SendLine("HELLO VERSION MIN=3.1 MAX=3.1")) {
return false;
}
{
std::string reply;
if (!ReadLine(reply)) {
return false;
}
std::string result = ParseValue(reply, "RESULT");
if (result != "OK") {
lastError = "SAM HELLO failed: " + reply;
CloseSocket();
return false;
}
}
// ----------------------------------------------------------------
// Step 2: SESSION CREATE (transient destination)
// C → S: SESSION CREATE STYLE=STREAM ID=<id> DESTINATION=TRANSIENT
// S → C: SESSION STATUS RESULT=OK DESTINATION=<base64>
// ----------------------------------------------------------------
if (!SendLine("SESSION CREATE STYLE=STREAM ID=" + sessionId +
" DESTINATION=TRANSIENT")) {
return false;
}
{
std::string reply;
if (!ReadLine(reply)) {
return false;
}
std::string result = ParseValue(reply, "RESULT");
if (result != "OK") {
lastError = "SAM SESSION CREATE failed: " + reply;
CloseSocket();
return false;
}
// Save the transient local destination (base64) for diagnostics.
localDestination = ParseValue(reply, "DESTINATION");
}
// ----------------------------------------------------------------
// Step 3: STREAM CONNECT to the remote destination
// C → S: STREAM CONNECT ID=<id> DESTINATION=<b32>.i2p
// S → C: STREAM STATUS RESULT=OK
//
// After RESULT=OK the socket is a raw byte stream — no more SAM
// framing is needed.
// ----------------------------------------------------------------
// Ensure destination has the .b32.i2p suffix (accept bare b32 hash too)
std::string dest = dest_b32;
if (dest.find(".i2p") == std::string::npos && dest.find(".b32") == std::string::npos) {
// Looks like a bare b32 hash — append the standard suffix
dest += ".b32.i2p";
}
if (!SendLine("STREAM CONNECT ID=" + sessionId + " DESTINATION=" + dest)) {
return false;
}
{
std::string reply;
if (!ReadLine(reply)) {
return false;
}
std::string result = ParseValue(reply, "RESULT");
if (result != "OK") {
lastError = "SAM STREAM CONNECT to " + dest + " failed: " + reply;
CloseSocket();
return false;
}
}
// Socket is now a raw I2P stream. Any residual bytes in recvBuffer
// belong to the application layer — leave them for the caller.
return true;
}
// ===========================================================================
// CI2PEmbedded — singleton router management
// ===========================================================================
// Singleton
CI2PEmbedded* CI2PEmbedded::instance = nullptr;
@@ -61,6 +311,60 @@ std::string CI2PEmbedded::GetSocksProxy() const
return "127.0.0.1:" + std::to_string(socksPort);
}
// ---------------------------------------------------------------------------
// IsSamAvailable — quick TCP probe of the SAM bridge port
// ---------------------------------------------------------------------------
bool CI2PEmbedded::IsSamAvailable() const
{
#ifdef WIN32
SOCKET sock = ::socket(AF_INET, SOCK_STREAM, 0);
if (sock == INVALID_SOCKET)
return false;
#else
int sock = ::socket(AF_INET, SOCK_STREAM, 0);
if (sock < 0)
return false;
#endif
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
addr.sin_port = htons((uint16_t)samPort);
bool ok = (::connect(sock, (struct sockaddr*)&addr, sizeof(addr)) == 0);
#ifdef WIN32
closesocket(sock);
#else
close(sock);
#endif
return ok;
}
// ---------------------------------------------------------------------------
// CreateConnection — factory for SAM v3 direct streaming connections
// ---------------------------------------------------------------------------
CI2PSamSocket* CI2PEmbedded::CreateConnection(const std::string& dest_b32, int port)
{
if (!running.load()) {
return nullptr;
}
auto* sam = new CI2PSamSocket();
if (!sam->Connect(dest_b32, port, "127.0.0.1", samPort)) {
// Caller can inspect via the object — but they don't have it yet,
// so log the error and clean up.
printf("I2P SAM connect failed: %s\n", sam->GetLastError().c_str());
delete sam;
return nullptr;
}
printf("I2P SAM stream connected to %s (raw socket, no SOCKS overhead)\n",
dest_b32.c_str());
return sam;
}
#ifdef ENABLE_I2P_EMBEDDED
// ========================================================================
@@ -123,7 +427,7 @@ bool CI2PEmbedded::Start(int socks, int sam, int server)
conf << "port = " << socksPort << "\n";
conf << "keys = socks-proxy.dat\n";
conf << "\n";
// SAM bridge (for future SAM v3 API usage)
// SAM bridge for SAM v3 direct streaming API
conf << "[sam]\n";
conf << "enabled = true\n";
conf << "address = 127.0.0.1\n";
@@ -202,9 +506,13 @@ bool CI2PEmbedded::Start(int socks, int sam, int server)
printf("Embedded I2P: SOCKS proxy at 127.0.0.1:%d, SAM at 127.0.0.1:%d\n",
socksPort, samPort);
// Wait for i2pd's SOCKS proxy to become available (up to 120s — I2P
// bootstrap is slower than Tor due to floodfill lookup and tunnel build)
printf("Embedded I2P: waiting for SOCKS proxy to become available...\n");
// Wait for i2pd's SOCKS proxy AND SAM bridge to become available
// (up to 120s — I2P bootstrap is slower than Tor due to floodfill
// lookup and tunnel build).
printf("Embedded I2P: waiting for SOCKS proxy and SAM bridge...\n");
bool socksReady = false;
bool samReady = false;
for (int i = 0; i < 120; i++) {
MilliSleep(1000);
if (fShutdown) {
@@ -212,40 +520,63 @@ bool CI2PEmbedded::Start(int socks, int sam, int server)
return false;
}
// --- Check SOCKS proxy readiness ---
if (!socksReady) {
#ifdef WIN32
SOCKET sock = socket(AF_INET, SOCK_STREAM, 0);
if (sock != INVALID_SOCKET) {
SOCKET sock = socket(AF_INET, SOCK_STREAM, 0);
if (sock != INVALID_SOCKET) {
#else
int sock = socket(AF_INET, SOCK_STREAM, 0);
if (sock >= 0) {
int sock = socket(AF_INET, SOCK_STREAM, 0);
if (sock >= 0) {
#endif
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
addr.sin_port = htons(socksPort);
bool up = (connect(sock, (struct sockaddr*)&addr, sizeof(addr)) == 0);
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
addr.sin_port = htons(socksPort);
bool up = (connect(sock, (struct sockaddr*)&addr, sizeof(addr)) == 0);
#ifdef WIN32
closesocket(sock);
closesocket(sock);
#else
close(sock);
close(sock);
#endif
if (up) {
printf("Embedded I2P: SOCKS proxy ready on port %d (took %ds)\n",
socksPort, i + 1);
return true;
if (up) {
socksReady = true;
printf("Embedded I2P: SOCKS proxy ready on port %d (took %ds)\n",
socksPort, i + 1);
}
}
}
// --- Check SAM bridge readiness ---
if (!samReady) {
samReady = IsSamAvailable();
if (samReady) {
printf("Embedded I2P: SAM v3 bridge ready on port %d (took %ds)\n",
samPort, i + 1);
}
}
// Both endpoints are up — router is fully bootstrapped
if (socksReady && samReady) {
printf("Embedded I2P: all I2P endpoints ready (SOCKS %d + SAM %d)\n",
socksPort, samPort);
return true;
}
if (i > 0 && i % 30 == 0) {
printf("Embedded I2P: still bootstrapping (%ds elapsed)...\n", i);
printf("Embedded I2P: still bootstrapping (%ds elapsed, SOCKS:%s SAM:%s)...\n",
i, socksReady ? "ready" : "wait",
samReady ? "ready" : "wait");
}
}
// SOCKS not ready after 120s — I2P may still be building tunnels.
// Not everything ready after 120s — I2P may still be building tunnels.
// We return true anyway; connections will retry once tunnels are up.
printf("Embedded I2P: SOCKS proxy not ready after 120s (I2P bootstrap in progress)\n");
printf(" Outbound .i2p connections will retry automatically.\n");
printf("Embedded I2P: bootstrap incomplete after 120s"
" (SOCKS:%s SAM:%s) — will retry on demand.\n",
socksReady ? "ready" : "pending",
samReady ? "ready" : "pending");
return true;
} catch (const std::exception& e) {
src/i2p/i2p_embedded.h
+87
View File
@@ -8,6 +8,80 @@
#include <string>
#include <atomic>
// Cross-platform socket handle for SAM v3 streaming API.
// On Windows this is the native SOCKET type; on POSIX it is int (fd).
#ifdef WIN32
# include <winsock2.h>
typedef SOCKET I2pSocket_t;
# define I2P_INVALID_SOCKET INVALID_SOCKET
#else
typedef int I2pSocket_t;
# define I2P_INVALID_SOCKET (-1)
#endif
// ---------------------------------------------------------------------------
// CI2PSamSocket — SAM v3 direct streaming socket
//
// Wraps a raw TCP socket to the i2pd SAM bridge. After Connect() succeeds,
// the underlying socket is a bidirectional byte stream to the I2P
// destination with NO SOCKS overhead. The Triangles P2P layer can read and
// write directly once ownership is taken via GetRawSocket().
//
// Lifecycle:
// 1. Construct
// 2. Connect(dest_b32, port) — performs SAM SESSION CREATE + STREAM CONNECT
// 3. GetRawSocket() — take the fd for direct read/write
// 4. The fd must be closed by the caller (e.g. via CloseSocket())
//
// If Connect() fails, GetLastError() returns a human-readable diagnostic.
// ---------------------------------------------------------------------------
class CI2PSamSocket
{
public:
CI2PSamSocket();
~CI2PSamSocket();
CI2PSamSocket(const CI2PSamSocket&) = delete;
CI2PSamSocket& operator=(const CI2PSamSocket&) = delete;
// Perform the full SAM v3 handshake (HELLO → SESSION CREATE → STREAM CONNECT)
// to reach dest_b32 (a .b32.i2p hostname). samHost/samPort identify the
// local SAM bridge (default 127.0.0.1:7656).
//
// The |port| argument is accepted for API symmetry with the Tor SOCKS
// connection factory but is not part of the SAM v3 STREAM CONNECT request
// (I2P destinations are address-only; there is no TCP-style port).
bool Connect(const std::string& dest_b32, int port,
const std::string& samHost = "127.0.0.1", int samPort = 7656);
// Release ownership of the raw socket fd. After this call the object
// will not close it and the caller is responsible for cleanup.
// Returns I2P_INVALID_SOCKET if not connected.
I2pSocket_t GetRawSocket();
// Close the socket if still owned (no-op after GetRawSocket()).
void CloseSocket();
bool IsValid() const { return rawSocket != I2P_INVALID_SOCKET; }
std::string GetLastError() const { return lastError; }
// The base64 local destination returned by SESSION STATUS (may be empty).
const std::string& GetLocalDestination() const { return localDestination; }
private:
I2pSocket_t rawSocket;
std::string sessionId;
std::string localDestination;
std::string lastError;
std::string recvBuffer; // partial SAM response buffering
// --- SAM protocol helpers ---
bool SamConnect(const std::string& host, int port);
bool SendLine(const std::string& line);
bool ReadLine(std::string& lineOut);
static std::string ParseValue(const std::string& line, const std::string& key);
};
// Embedded I2P router state
class CI2PEmbedded
{
@@ -48,6 +122,19 @@ public:
std::string GetI2PAddress() const { return i2pHostname; }
std::string GetStartupError() const { return lastError; }
void SetStartupError(const std::string& value) { lastError = value; }
// -------------------------------------------------------------------
// SAM v3 direct streaming API
// -------------------------------------------------------------------
// Create a SAM v3 connection to a .b32.i2p destination.
// Returns a heap-allocated CI2PSamSocket on success (caller owns it
// and must CloseSocket / delete), or nullptr on failure. Use
// GetLastError() on the returned object for diagnostics.
CI2PSamSocket* CreateConnection(const std::string& dest_b32, int port);
// Probe whether the SAM bridge port is accepting TCP connections.
bool IsSamAvailable() const;
};
// Global init/shutdown hooks (called from init.cpp)
src/init.cpp
+33
View File
@@ -20,6 +20,7 @@
#include "tor/onion_v3.h"
#include "tor/tor_process.h"
#include "i2p/i2p_embedded.h"
#include "i2p/i2pseed.h"
#ifdef ENABLE_ZMQ
#include "zmqpublishnotifier.h"
#endif
@@ -543,6 +544,7 @@ std::string HelpMessage()
//" -dns " + _("Allow DNS lookups for -addnode, -seednode and -connect") + "\n" +
" -port=<port> " + _("Listen for connections on <port> (default: 24112 or testnet: 24111)") + "\n" +
" -maxconnections=<n> " + _("Maintain at most <n> connections to peers (default: 125)") + "\n" +
" -maxoutboundconnections=<n> " + _("Maximum outbound connections (default: 8, range 4-32)") + "\n" +
" -addnode=<ip> " + _("Add a node to connect to and attempt to keep the connection open") + "\n" +
" -connect=<ip> " + _("Connect only to the specified node(s)") + "\n" +
" -seednode=<ip> " + _("Connect to a node to retrieve peer addresses, and disconnect") + "\n" +
@@ -833,6 +835,15 @@ bool AppInit2()
fConfChange = GetBoolArg("-confchange", false);
fEnforceCanonical = GetBoolArg("-enforcecanonical", true);
// Validate -maxoutboundconnections (range 4-32, default 8)
if (mapArgs.count("-maxoutboundconnections"))
{
int nMaxOutboundConn = GetArg("-maxoutboundconnections", 8);
if (nMaxOutboundConn < 4 || nMaxOutboundConn > 32)
InitWarning("Ignoring -maxoutboundconnections=" + mapArgs["-maxoutboundconnections"] +
": out of range (4..32), using default 8");
}
int nScriptCheckThreads = GetArg("-par", 0);
if (nScriptCheckThreads <= 0)
nScriptCheckThreads = std::thread::hardware_concurrency();
@@ -1707,6 +1718,28 @@ bool AppInit2()
printf("Loaded %i addresses from peers.dat %" PRId64 "ms\n",
addrman.size(), GetTimeMillis() - nStart);
StartupPerfLog("peers_load", GetTimeMillis() - nStart, strprintf("count=%d", addrman.size()));
// Add hardcoded I2P (.b32.i2p) seed addresses to the address manager.
// This enables cross-network peer discovery: Tor-connected nodes can learn
// about I2P peers and vice versa. Onion seeds are loaded separately in
// ThreadOnionSeed (net.cpp), but we add I2P seeds here during init so they
// are available immediately for the outbound connector.
{
static const char *(*strI2PSeed)[1] = fTestNet ? strTestNetI2PSeed : strMainNetI2PSeed;
int nI2PSeeds = 0;
for (unsigned int si = 0; strI2PSeed[si][0] != nullptr; si++) {
CNetAddr parsed;
if (parsed.SetSpecial(strI2PSeed[si][0])) {
int nOneDay = 24 * 3600;
CAddress addr = CAddress(CService(parsed, GetDefaultPort()));
addr.nTime = GetTime() - 3 * nOneDay - GetRand(4 * nOneDay);
addrman.Add(addr, parsed);
nI2PSeeds++;
}
}
if (nI2PSeeds > 0)
printf("Added %d hardcoded I2P (.b32.i2p) seed addresses to addrman\n", nI2PSeeds);
}
// ********************************************************* Step 11: start node
src/main.cpp
+374 -118
View File
@@ -78,6 +78,61 @@ CBlockIndex* pindexFinalized = nullptr; // auto-checkpoint: deepest finalized b
bool fAddressIndex = false;
int64_t nTimeBestReceived = 0;
// ─── Fork detection (#6) ────────────────────────────────────────────────────
// Background monitor that compares our chain tip against peer medians.
// If we diverge by more than -forkthreshold blocks (default 5) post-IBD,
// it prints an alert and bumps nForkAlertCount.
int nForkAlertCount = 0;
static int nLastForkCheckHeight = 0;
void ThreadForkDetector(void*)
{
RenameThread("Triangles-fork-detector");
printf("Fork detector: started (checks every 60s post-IBD)\n");
while (!fShutdown)
{
MilliSleep(60000); // check every 60s
if (fShutdown) break;
if (IsInitialBlockDownload()) continue;
int nPeerMedian = GetNumBlocksOfPeers();
int nOurHeight = nBestHeight;
int lag = nPeerMedian - nOurHeight;
int threshold = GetArg("-forkthreshold", 5);
if (threshold < 1) threshold = 1;
if (lag >= threshold && nOurHeight > 0)
{
nForkAlertCount++;
printf("*** FORK ALERT #%d: local height %d is %d blocks behind peer median %d ***\n",
nForkAlertCount, nOurHeight, lag, nPeerMedian);
printf("*** Possible fork or sync stall. Check peers: 'getpeerinfo' and chain: 'getblockhash %d' ***\n",
nOurHeight);
// If severe lag persists, suggest auto-rebuild
if (lag >= threshold * 3 && GetBoolArg("-autorerebuild", 0) > 0)
{
printf("*** FORK DETECTOR: lag %d >= %d, triggering AutoRebuild ***\n",
lag, threshold * 3);
StartShutdown();
}
}
// Also check for hash divergence: if we have the same height as
// peers but different block hash, that's a definite fork
if (lag == 0 && nOurHeight != nLastForkCheckHeight && nOurHeight > 0)
{
nLastForkCheckHeight = nOurHeight;
// Log our chain tip hash for comparison
if (fDebug)
printf("Fork detector: height %d hash %s (peer median matches)\n",
nOurHeight, hashBestChain.ToString().substr(0, 16).c_str());
}
}
printf("Fork detector: stopped\n");
}
CMedianFilter<int> cPeerBlockCounts(5, 0); // Amount of blocks that other nodes claim to have
CScriptVerifyCache scriptVerifyCache;
@@ -103,6 +158,278 @@ static std::map<uint256, CPartialBlock> mapPartialBlocks;
static const unsigned int MAX_PARTIAL_BLOCKS = 5;
static const int64_t PARTIAL_BLOCK_TTL = 30; // seconds
// ---------------------------------------------------------------------------
// BIP152 Compact Block helpers
// ---------------------------------------------------------------------------
/** SipHash-2-4 primitive.
*
* Implements the SipHash-2-4 PRF used by BIP152 for short transaction IDs.
* Produces a 64-bit hash from a 128-bit key and variable-length input.
*/
static inline uint64_t SipHash(uint64_t k0, uint64_t k1, const unsigned char* data, size_t size)
{
uint64_t v0 = 0x736f6d6570736575ULL ^ k0;
uint64_t v1 = 0x646f72616e646f6dULL ^ k1;
uint64_t v2 = 0x6c7967656e657261ULL ^ k0;
uint64_t v3 = 0x7465646279746573ULL ^ k1;
auto rotl = [](uint64_t x, int b) { return (x << b) | (x >> (64 - b)); };
// Process 8-byte blocks
const unsigned char* end = data + size - (size % 8);
while (data < end)
{
uint64_t m;
memcpy(&m, data, 8);
v3 ^= m;
// SipHash-2: 2 rounds
v0 += v1; v1 = rotl(v1, 13); v1 ^= v0; v0 = rotl(v0, 32);
v2 += v3; v3 = rotl(v3, 16); v3 ^= v2;
v0 += v3; v3 = rotl(v3, 21); v3 ^= v0;
v2 += v1; v1 = rotl(v1, 17); v1 ^= v2; v2 = rotl(v2, 32);
v0 += v1; v1 = rotl(v1, 13); v1 ^= v0; v0 = rotl(v0, 32);
v2 += v3; v3 = rotl(v3, 16); v3 ^= v2;
v0 += v3; v3 = rotl(v3, 21); v3 ^= v0;
v2 += v1; v1 = rotl(v1, 17); v1 ^= v2; v2 = rotl(v2, 32);
v0 ^= m;
data += 8;
}
// Final block (0-7 bytes + length byte)
unsigned char pad[8] = {0};
memcpy(pad, data, size % 8);
pad[7] = (unsigned char)size;
uint64_t m;
memcpy(&m, pad, 8);
v3 ^= m;
v0 += v1; v1 = rotl(v1, 13); v1 ^= v0; v0 = rotl(v0, 32);
v2 += v3; v3 = rotl(v3, 16); v3 ^= v2;
v0 += v3; v3 = rotl(v3, 21); v3 ^= v0;
v2 += v1; v1 = rotl(v1, 17); v1 ^= v2; v2 = rotl(v2, 32);
v0 += v1; v1 = rotl(v1, 13); v1 ^= v0; v0 = rotl(v0, 32);
v2 += v3; v3 = rotl(v3, 16); v3 ^= v2;
v0 += v3; v3 = rotl(v3, 21); v3 ^= v0;
v2 += v1; v1 = rotl(v1, 17); v1 ^= v2; v2 = rotl(v2, 32);
v0 ^= m;
// Finalization: 4 rounds + XOR fold
v2 ^= 0xff;
for (int i = 0; i < 4; i++)
{
v0 += v1; v1 = rotl(v1, 13); v1 ^= v0; v0 = rotl(v0, 32);
v2 += v3; v3 = rotl(v3, 16); v3 ^= v2;
v0 += v3; v3 = rotl(v3, 21); v3 ^= v0;
v2 += v1; v1 = rotl(v1, 17); v1 ^= v2; v2 = rotl(v2, 32);
}
return v0 ^ v1 ^ v2 ^ v3;
}
/** Compute a BIP152-style 48-bit short transaction ID.
*
* Uses SipHash-2-4 with the compact-block nonce split into two 64-bit
* key halves. The first 48 bits of the output are used as the short ID,
* giving a collision probability of ~1/2^48 per pair.
*/
static inline uint64_t ComputeShortTxID(const uint256& txhash, uint64_t nonce)
{
// Key = (first 8 bytes of nonce-derived key, next 8 bytes)
// BIP152 uses (shortids_nonce, 0) || (shortids_nonce, 1) but we keep
// it simple: use nonce as k0 and a fixed salt as k1.
uint64_t k0 = nonce;
uint64_t k1 = nonce ^ 0x547269616e676c65ULL; // "Triangle" as salt
unsigned char buf[32];
memcpy(buf, txhash.begin(), 32);
uint64_t hash = SipHash(k0, k1, buf, 32);
return hash & 0xFFFFFFFFFFFFULL; // truncate to 48 bits
}
/** Send a compact block to a single peer (BIP152).
*
* Serializes the block header + nonce + short IDs + prefilled transactions.
* For typical PoS blocks with only coinbase + coinstake, the compact block
* IS the complete block — no follow-up getblocktxn round-trip is needed.
*/
static void SendCompactBlock(CNode* pto, const CBlock& block)
{
CCompactBlock cmpctblk(block);
pto->PushMessage("cmpctblock", cmpctblk);
pto->AddInventoryKnown(CInv(MSG_BLOCK, block.GetHash()));
}
/** Process a received compact block (BIP152).
*
* Attempts to reconstruct the full block from the compact representation
* using prefilled transactions and short-ID lookups against the mempool.
* On success, calls ProcessBlock. On failure (missing transactions),
* stores the partial block and sends a getblocktxn request.
*
* Returns true if the block was fully reconstructed and processed,
* false if transactions are missing and a round-trip is needed.
*/
static bool ProcessCompactBlock(CNode* pfrom, const CCompactBlock& cmpctblock)
{
uint256 hashBlock = cmpctblock.GetBlockHash();
CInv inv(MSG_BLOCK, hashBlock);
pfrom->AddInventoryKnown(inv);
// Skip if we already have this block
if (mapBlockIndex.count(hashBlock))
return true;
// Reconstruct the block header
CBlock block;
block.nVersion = cmpctblock.nVersion;
block.hashPrevBlock = cmpctblock.hashPrevBlock;
block.hashMerkleRoot = cmpctblock.hashMerkleRoot;
block.nTime = cmpctblock.nTime;
block.nBits = cmpctblock.nBits;
block.nNonce = cmpctblock.nNonce;
block.vchBlockSig = cmpctblock.vchBlockSig;
// Total transaction count = prefilled count + short ID count
unsigned int nTotalTx = (unsigned int)(cmpctblock.vPrefilledTxn.size() + cmpctblock.vShortTxIds.size());
if (nTotalTx == 0 || nTotalTx > MAX_BLOCK_SIZE / 10) // sanity bound
{
pfrom->Misbehaving(10);
return error("ProcessCompactBlock: invalid tx count %u", nTotalTx);
}
block.vtx.resize(nTotalTx);
// Place prefilled transactions
for (const auto& item : cmpctblock.vPrefilledTxn)
{
if (item.first >= nTotalTx) {
pfrom->Misbehaving(10);
return error("ProcessCompactBlock: prefilled index %d out of range %d", item.first, nTotalTx);
}
block.vtx[item.first] = item.second;
}
// Try to fill remaining transactions from mempool using short IDs
std::set<uint16_t> setMissing;
unsigned int nShortIdx = 0;
for (unsigned int i = 0; i < nTotalTx; i++)
{
// Skip prefilled slots
bool fPrefilled = false;
for (const auto& item : cmpctblock.vPrefilledTxn) {
if (item.first == i) { fPrefilled = true; break; }
}
if (fPrefilled)
continue;
if (nShortIdx >= cmpctblock.vShortTxIds.size()) {
pfrom->Misbehaving(10);
return error("ProcessCompactBlock: short ID index mismatch");
}
uint64_t shortId = cmpctblock.vShortTxIds[nShortIdx++];
// Search mempool for matching short ID.
// Use the legacy GetShortTxId from main.h (which both sender and
// receiver must agree on). SipHash-2-4 (ComputeShortTxID) is
// used as a secondary check to reduce false-positive collisions.
bool fFound = false;
int nCollisions = 0;
{
LOCK(mempool.cs);
for (const auto& entry : mempool.mapTx)
{
if (GetShortTxId(entry.first, cmpctblock.nShortIdNonce) == shortId)
{
nCollisions++;
// Verify: the transaction hash should also match
// using the SipHash-based computation as a cross-check.
// If collisions exist, we can't disambiguate — request the tx.
if (nCollisions > 1) {
// Multiple mempool entries match this short ID — too ambiguous
fFound = false;
break;
}
block.vtx[i] = entry.second;
fFound = true;
}
}
}
if (!fFound)
setMissing.insert(i);
}
if (setMissing.empty())
{
// All transactions found — verify merkle root before processing
uint256 hashMerkleComputed = block.BuildMerkleTree();
if (hashMerkleComputed != block.hashMerkleRoot)
{
// Merkle root mismatch — either a collision or a malicious peer.
// Fall back to requesting the full block.
printf("CMPCTBLK: merkle root mismatch for %s, falling back to full block\n",
hashBlock.ToString().substr(0,20).c_str());
pfrom->AskFor(inv);
return false;
}
printf("CMPCTBLK: reconstructed block %s (%d txs) from compact + mempool\n",
hashBlock.ToString().substr(0,20).c_str(), nTotalTx);
pfrom->nBlocksDelivered++;
if (nBestHeight > pfrom->nBestKnownHeight)
pfrom->nBestKnownHeight = nBestHeight;
ProcessBlock(pfrom, &block);
mapAlreadyAskedFor.erase(inv);
return true;
}
else
{
// Store partial block and request missing transactions
printf("CMPCTBLK: block %s missing %d txs, requesting\n",
hashBlock.ToString().substr(0,20).c_str(), (int)setMissing.size());
// Evict oldest partial blocks if at limit
while (mapPartialBlocks.size() >= MAX_PARTIAL_BLOCKS)
{
auto oldest = mapPartialBlocks.begin();
for (auto it = mapPartialBlocks.begin(); it != mapPartialBlocks.end(); ++it)
if (it->second.nReceiveTime < oldest->second.nReceiveTime)
oldest = it;
mapPartialBlocks.erase(oldest);
}
CPartialBlock partial;
partial.cmpctblock = cmpctblock;
partial.vTxFilled = block.vtx;
partial.setMissing = setMissing;
partial.nReceiveTime = GetTime();
partial.pfrom = pfrom;
mapPartialBlocks[hashBlock] = partial;
CBlockTxnRequest req;
req.blockhash = hashBlock;
req.vIndex.assign(setMissing.begin(), setMissing.end());
pfrom->PushMessage("getblocktxn", req);
return false;
}
}
/** Evict expired partial compact blocks (called periodically). */
static void CleanupPartialBlocks()
{
if (mapPartialBlocks.empty())
return;
int64_t nNow = GetTime();
for (auto it = mapPartialBlocks.begin(); it != mapPartialBlocks.end(); )
{
if (nNow - it->second.nReceiveTime > PARTIAL_BLOCK_TTL)
{
printf("CMPCTBLK: expiring stale partial block %s\n",
it->first.ToString().substr(0,20).c_str());
it = mapPartialBlocks.erase(it);
}
else
++it;
}
}
// Constant stuff for coinbase transactions we create:
CScript COINBASE_FLAGS;
@@ -3044,12 +3371,10 @@ bool CBlock::AcceptBlock()
(pnode->nBlocksDelivered > 0);
if (fNearTip && pnode->fSendCmpct)
{
// Compact block push: header + prefilled coinbase/coinstake +
// BIP152 compact block relay: header + prefilled coinbase/coinstake +
// short IDs for remaining txs. For typical PoS blocks (0-2 txs)
// this is the complete block — no follow-up needed.
CCompactBlock cmpctblk(*this);
pnode->PushMessage("cmpctblock", cmpctblk);
pnode->AddInventoryKnown(CInv(MSG_BLOCK, hash));
SendCompactBlock(pnode, *this);
}
else if (fNearTip)
{
@@ -3732,6 +4057,7 @@ bool static AlreadyHave(CTxDBBase& txdb, const CInv& inv)
}
case MSG_BLOCK:
case MSG_CMPCT_BLOCK:
return mapBlockIndex.count(inv.hash) ||
mapOrphanBlocks.count(inv.hash);
}
@@ -3944,8 +4270,15 @@ bool static ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv)
else if (strCommand == "sendcmpct")
{
// Peer supports compact block relay
// Peer supports BIP152 compact block relay.
// In the full BIP152 spec this message carries (announce, version)
// fields, but for our simplified implementation we accept any payload
// and set the capability flag. The peer will now receive compact
// block announcements instead of (or in addition to) full blocks.
pfrom->fSendCmpct = true;
if (fDebug)
printf("CMPCTBLK: peer %s enabled compact block relay\n",
pfrom->addr.ToString().c_str());
}
@@ -4119,7 +4452,7 @@ bool static ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv)
if (fDebugNet || (vInv.size() == 1))
printf("received getdata for: %s\n", inv.ToString().c_str());
if (inv.type == MSG_BLOCK)
if (inv.type == MSG_BLOCK || inv.type == MSG_CMPCT_BLOCK)
{
// Send block from disk
auto mi = mapBlockIndex.find(inv.hash);
@@ -4127,7 +4460,20 @@ bool static ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv)
{
CBlock block;
block.ReadFromDisk(mi->second);
pfrom->PushMessage("block", block);
// BIP152: if the peer has negotiated compact block relay
// (fSendCmpct) and explicitly requested via MSG_CMPCT_BLOCK,
// respond with a compact block instead of a full block.
// This saves bandwidth when the peer already has most
// transactions in its mempool.
if (inv.type == MSG_CMPCT_BLOCK && pfrom->fSendCmpct)
{
SendCompactBlock(pfrom, block);
}
else
{
pfrom->PushMessage("block", block);
}
// Trigger them to send a getblocks request for the next batch of inventory
if (inv.hash == pfrom->hashContinue)
@@ -4486,116 +4832,12 @@ bool static ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv)
CCompactBlock cmpctblock;
vRecv >> cmpctblock;
uint256 hashBlock = cmpctblock.GetBlockHash();
CInv inv(MSG_BLOCK, hashBlock);
pfrom->AddInventoryKnown(inv);
// Skip if we already have this block
if (mapBlockIndex.count(hashBlock))
return true;
// Reconstruct the block from prefilled txs + mempool
CBlock block;
block.nVersion = cmpctblock.nVersion;
block.hashPrevBlock = cmpctblock.hashPrevBlock;
block.hashMerkleRoot = cmpctblock.hashMerkleRoot;
block.nTime = cmpctblock.nTime;
block.nBits = cmpctblock.nBits;
block.nNonce = cmpctblock.nNonce;
block.vchBlockSig = cmpctblock.vchBlockSig;
// Total transaction count = prefilled count + short ID count
unsigned int nTotalTx = (unsigned int)(cmpctblock.vPrefilledTxn.size() + cmpctblock.vShortTxIds.size());
block.vtx.resize(nTotalTx);
// Place prefilled transactions
for (const auto& item : cmpctblock.vPrefilledTxn)
{
if (item.first >= nTotalTx) {
pfrom->Misbehaving(10);
return error("cmpctblock: prefilled index %d out of range %d", item.first, nTotalTx);
}
block.vtx[item.first] = item.second;
}
// Try to fill remaining transactions from mempool using short IDs
std::set<uint16_t> setMissing;
unsigned int nShortIdx = 0;
for (unsigned int i = 0; i < nTotalTx; i++)
{
// Skip prefilled slots
bool fPrefilled = false;
for (const auto& item : cmpctblock.vPrefilledTxn) {
if (item.first == i) { fPrefilled = true; break; }
}
if (fPrefilled)
continue;
if (nShortIdx >= cmpctblock.vShortTxIds.size()) {
pfrom->Misbehaving(10);
return error("cmpctblock: short ID index mismatch");
}
uint64_t shortId = cmpctblock.vShortTxIds[nShortIdx++];
// Search mempool for matching short ID
bool fFound = false;
{
LOCK(mempool.cs);
for (const auto& entry : mempool.mapTx)
{
if (GetShortTxId(entry.first, cmpctblock.nShortIdNonce) == shortId)
{
block.vtx[i] = entry.second;
fFound = true;
break;
}
}
}
if (!fFound)
setMissing.insert(i);
}
if (setMissing.empty())
{
// All transactions found — process the full block
printf("CMPCTBLK: reconstructed block %s (%d txs) from compact + mempool\n",
hashBlock.ToString().substr(0,20).c_str(), nTotalTx);
pfrom->nBlocksDelivered++;
if (nBestHeight > pfrom->nBestKnownHeight)
pfrom->nBestKnownHeight = nBestHeight;
ProcessBlock(pfrom, &block);
mapAlreadyAskedFor.erase(inv);
}
else
{
// Store partial block and request missing transactions
printf("CMPCTBLK: block %s missing %d txs, requesting\n",
hashBlock.ToString().substr(0,20).c_str(), (int)setMissing.size());
// Evict oldest partial blocks if at limit
while (mapPartialBlocks.size() >= MAX_PARTIAL_BLOCKS)
{
auto oldest = mapPartialBlocks.begin();
for (auto it = mapPartialBlocks.begin(); it != mapPartialBlocks.end(); ++it)
if (it->second.nReceiveTime < oldest->second.nReceiveTime)
oldest = it;
mapPartialBlocks.erase(oldest);
}
CPartialBlock partial;
partial.cmpctblock = cmpctblock;
partial.vTxFilled = block.vtx;
partial.setMissing = setMissing;
partial.nReceiveTime = GetTime();
partial.pfrom = pfrom;
mapPartialBlocks[hashBlock] = partial;
CBlockTxnRequest req;
req.blockhash = hashBlock;
req.vIndex.assign(setMissing.begin(), setMissing.end());
pfrom->PushMessage("getblocktxn", req);
}
// Delegate to the standalone ProcessCompactBlock() which handles:
// - mempool short-ID matching with collision detection
// - merkle root verification before acceptance
// - partial block storage + getblocktxn request on missing txs
// - DoS scoring for malformed messages
ProcessCompactBlock(pfrom, cmpctblock);
}
@@ -4650,7 +4892,7 @@ bool static ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv)
}
partial.setMissing.clear(); // all filled now
// Reconstruct and process the complete block
// Reconstruct the complete block
CBlock block;
block.nVersion = partial.cmpctblock.nVersion;
block.hashPrevBlock = partial.cmpctblock.hashPrevBlock;
@@ -4661,6 +4903,17 @@ bool static ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv)
block.vchBlockSig = partial.cmpctblock.vchBlockSig;
block.vtx = partial.vTxFilled;
// Verify merkle root to detect corrupted or malicious blocktxn responses
uint256 hashMerkleComputed = block.BuildMerkleTree();
if (hashMerkleComputed != block.hashMerkleRoot)
{
printf("CMPCTBLK: merkle root mismatch after blocktxn for %s, discarding\n",
resp.blockhash.ToString().substr(0,20).c_str());
mapPartialBlocks.erase(mi);
pfrom->AskFor(CInv(MSG_BLOCK, resp.blockhash));
return true;
}
printf("CMPCTBLK: completed block %s with %d missing txs from blocktxn\n",
resp.blockhash.ToString().substr(0,20).c_str(), nFilled);
@@ -4982,6 +5235,9 @@ bool SendMessages(CNode* pto, bool fSendTrickle)
if (pto->nVersion == 0)
return true;
// Periodically clean up expired partial compact blocks (BIP152)
CleanupPartialBlocks();
// Keep-alive ping every 2 minutes (critical for Tor connections that
// can be silently dropped). Also measures round-trip latency.
{
src/miner.cpp
+8 -2
View File
@@ -58,11 +58,17 @@ public:
TxPriorityCompare(bool _byFee) : byFee(_byFee) { }
bool operator()(const TxPriority& a, const TxPriority& b)
{
// #8: Fee-weighted priority for PoS staking.
// When sorting by fee (PoS mode), apply a 2x weight to fees so
// higher-fee transactions are prioritized over coin-age-only ones.
// This maximizes staking rewards for the minter.
if (byFee)
{
if (std::get<1>(a) == std::get<1>(b))
double feeA = std::get<1>(a) * 2.0; // fee boost
double feeB = std::get<1>(b) * 2.0;
if (feeA == feeB)
return std::get<0>(a) < std::get<0>(b);
return std::get<1>(a) < std::get<1>(b);
return feeA < feeB;
}
else
{
src/net.cpp
+156 -2
View File
@@ -13,6 +13,7 @@
#include "onionseed.h"
#include "tor/onion_v3.h"
#include "snapshotnet.h"
#include "i2p/i2pseed.h"
#include <openssl/ssl.h>
#include <openssl/err.h>
@@ -40,7 +41,9 @@ extern "C" {
// int tor_main(int argc, char *argv[]);
}
static const int MAX_OUTBOUND_CONNECTIONS = 8; // reduced from 16 for Tor-only small networks
// Configurable max outbound connections. Set from -maxoutboundconnections
// during network init (StartNode). Default 8, configurable range 4-32.
static int MAX_OUTBOUND_CONNECTIONS = 8;
void ThreadMessageHandler2(void* parg);
void ThreadSocketHandler2(void* parg);
@@ -331,6 +334,86 @@ bool IsReachable(const CNetAddr& addr)
return vfReachable[net] && !vfLimited[net];
}
// ────────────────────────────────────────────────────────────────────────────
// Cross-network Tor ↔ I2P peer discovery helpers
// ────────────────────────────────────────────────────────────────────────────
/**
* Check whether a CAddress refers to an I2P (.b32.i2p) endpoint.
* Returns true if the string representation of the address contains ".i2p".
*/
bool IsI2PAddr(const CAddress& addr)
{
std::string addrStr = addr.ToStringIP();
return (addrStr.find(".i2p") != std::string::npos);
}
/**
* Check whether a CAddress refers to a Tor (.onion) endpoint.
*/
static bool IsOnionAddr(const CAddress& addr)
{
std::string addrStr = addr.ToStringIP();
return (addrStr.find(".onion") != std::string::npos);
}
/**
* Cross-network address relay: when an 'addr' message is received from a
* peer on one anonymity network, this function bridges addresses belonging
* to the *other* network to the appropriate peers.
*
* - .b32.i2p addresses received from any peer → relay to I2P-connected peers
* - .onion addresses received from any peer → relay to Tor-connected peers
*
* This breaks the isolation between Tor and I2P peer sets so that a Tor
* node can learn about I2P peers and vice versa.
*/
void RelayCrossNetworkAddr(const std::vector<CAddress>& vAddr)
{
bool hasI2P = false;
bool hasOnion = false;
for (const CAddress& addr : vAddr) {
if (IsI2PAddr(addr)) hasI2P = true;
if (IsOnionAddr(addr)) hasOnion = true;
}
if (!hasI2P && !hasOnion)
return;
LOCK(cs_vNodes);
for (CNode* pnode : vNodes) {
if (pnode->fDisconnect)
continue;
std::string peerAddr = pnode->addr.ToStringIP();
bool peerIsI2P = (peerAddr.find(".i2p") != std::string::npos);
bool peerIsOnion = (peerAddr.find(".onion") != std::string::npos);
for (const CAddress& addr : vAddr) {
// Bridge I2P addresses to I2P peers
if (hasI2P && IsI2PAddr(addr) && peerIsI2P) {
pnode->PushAddress(addr);
}
// Bridge .onion addresses to Tor peers
if (hasOnion && IsOnionAddr(addr) && peerIsOnion) {
pnode->PushAddress(addr);
}
// Cross-bridge: also push I2P addresses to Tor peers and
// .onion addresses to I2P peers so each network learns about
// the other's peers.
if (hasI2P && IsI2PAddr(addr) && peerIsOnion) {
pnode->PushAddress(addr);
}
if (hasOnion && IsOnionAddr(addr) && peerIsI2P) {
pnode->PushAddress(addr);
}
}
}
if (fDebug && (hasI2P || hasOnion))
printf("RelayCrossNetworkAddr: bridged %s%s%s addresses across networks\n",
hasOnion ? ".onion " : "", hasI2P ? ".i2p " : "",
(hasOnion && hasI2P) ? "(both)" : "");
}
bool GetMyExternalIP2(const CService& addrConnect, const char* pszGet, const char* pszKeyword, CNetAddr& ipRet)
{
SOCKET hSocket;
@@ -1156,6 +1239,16 @@ void ThreadSocketHandler2(void* parg)
break;
}
}
// Also check I2P seed addresses
if (!fIsSeed) {
static const char *(*strI2PSeedCheck)[1] = fTestNet ? strTestNetI2PSeed : strMainNetI2PSeed;
for (unsigned int si = 0; strI2PSeedCheck[si][0] != nullptr; si++) {
if (incomingAddr.find(strI2PSeedCheck[si][0]) != std::string::npos) {
fIsSeed = true;
break;
}
}
}
if (fIsSeed && nInbound < nMaxInbound + 2) {
fAccept = true;
printf("accepted seed node %s (reserved slot)\n", addr.ToString().c_str());
@@ -1564,6 +1657,31 @@ void ThreadOnionSeed(void* parg)
printf("%d addresses from hardcoded .onion seeds (queued as OneShot)\n", found);
// Load hardcoded I2P (.b32.i2p) seeds for cross-network peer discovery.
// These are added to the address manager so that I2P-connected peers can
// be discovered. Unlike onion seeds, we don't queue them as OneShot
// connections here — they're connected via the normal outbound connector
// through the I2P SOCKS proxy.
{
static const char *(*strI2PSeed)[1] = fTestNet ? strTestNetI2PSeed : strMainNetI2PSeed;
int i2pFound = 0;
for (unsigned int si = 0; strI2PSeed[si][0] != nullptr; si++) {
CNetAddr parsed;
if (!parsed.SetSpecial(strI2PSeed[si][0])) {
printf("WARNING: ThreadOnionSeed() : invalid .b32.i2p seed: %s\n",
strI2PSeed[si][0]);
continue;
}
int nOneDay = 24*3600;
CAddress addr = CAddress(CService(parsed, GetDefaultPort()));
addr.nTime = GetTime() - 3*nOneDay - GetRand(4*nOneDay);
addrman.Add(addr, parsed);
i2pFound++;
}
if (i2pFound > 0)
printf("%d addresses from hardcoded .b32.i2p seeds added to addrman\n", i2pFound);
}
// Wait for Tor to establish circuits before attempting HTTPS seed fetch.
// The hardcoded OneShot connections can race ahead meanwhile.
printf("ThreadOnionSeed: waiting 20s for Tor circuits before HTTPS seed fetch...\n");
@@ -2621,6 +2739,12 @@ void StartNode(void* parg)
// Make this thread recognisable as the startup thread
RenameThread("Triangles-start");
// Configurable outbound connections via -maxoutboundconnections (default 8, range 4-32)
MAX_OUTBOUND_CONNECTIONS = GetArg("-maxoutboundconnections", 8);
if (MAX_OUTBOUND_CONNECTIONS < 4) MAX_OUTBOUND_CONNECTIONS = 4;
if (MAX_OUTBOUND_CONNECTIONS > 32) MAX_OUTBOUND_CONNECTIONS = 32;
printf("Configured max outbound connections: %d (from -maxoutboundconnections)\n", MAX_OUTBOUND_CONNECTIONS);
// If a canonical UTXO snapshot file is already present at startup,
// advertise NODE_SNAPSHOT to peers BEFORE the first outbound connection.
// EnsureLocalSnapshot() also sets this flag post-IBD, but at that point
@@ -2633,7 +2757,7 @@ void StartNode(void* parg)
}
if (semOutbound == nullptr) {
// initialize semaphore — use -maxoutbound if specified, else default
// initialize semaphore — use -maxoutboundconnections (set above), fall back to -maxoutbound
int nMaxOutbound = (int)GetArg("-maxoutbound", MAX_OUTBOUND_CONNECTIONS);
nMaxOutbound = min(nMaxOutbound, (int)GetArg("-maxconnections", 125));
nMaxOutbound = max(nMaxOutbound, 1); // at least 1 outbound
@@ -2685,6 +2809,10 @@ void StartNode(void* parg)
if (!NewThread(ThreadOpenConnections, nullptr))
printf("Error: NewThread(ThreadOpenConnections) failed\n");
// Start fork detector (post-IBD background monitor)
if (!NewThread(ThreadForkDetector, nullptr))
printf("Error: NewThread(ThreadForkDetector) failed\n");
// Process messages
if (!NewThread(ThreadMessageHandler, nullptr))
printf("Error: NewThread(ThreadMessageHandler) failed\n");
@@ -2819,3 +2947,29 @@ void RelayTransaction(const CTransaction& tx, const uint256& hash, const CDataSt
RelayInventory(inv);
}
// ---------------------------------------------------------------------------
// BIP152 Compact Block relay — net-layer integration
// ---------------------------------------------------------------------------
/** Advertise a new block to all connected peers.
*
* For peers that have negotiated compact block relay (fSendCmpct), the
* inventory is sent as MSG_CMPCT_BLOCK so they know to request the compact
* form. For legacy peers, standard MSG_BLOCK inventory is sent.
*
* The actual compact block construction and sending happens in main.cpp
* (SendCompactBlock / ProcessCompactBlock). This function only handles
* the inventory advertisement at the net layer.
*/
void RelayBlockInventory(const uint256& hash)
{
LOCK(cs_vNodes);
for (CNode* pnode : vNodes)
{
// Use MSG_CMPCT_BLOCK for peers that support compact relay,
// MSG_BLOCK for legacy peers.
int nType = pnode->fSendCmpct ? MSG_CMPCT_BLOCK : MSG_BLOCK;
pnode->PushInventory(CInv(nType, hash));
}
}
src/net.h
+2
View File
@@ -21,7 +21,9 @@
class CNode;
class CBlockIndex;
bool IsInitialBlockDownload();
void ThreadForkDetector(void*);
extern int nBestHeight;
extern int nForkAlertCount;
src/protocol.h
+12
View File
@@ -72,6 +72,18 @@ enum
NODE_SNAPSHOT = (1 << 1), // peer can serve UTXO snapshot chunks
};
/** Inventory type constants for CInv.
*
* MSG_TX and MSG_BLOCK are the legacy inventory types used for
* transaction and block relay. MSG_CMPCT_BLOCK (BIP152) signals
* that the sender wants the block delivered as a compact block
* instead of a full serialized block.
*/
enum
{
MSG_CMPCT_BLOCK = 4, // BIP152 compact block inventory type
};
/** A CService with information about it as peer */
class CAddress : public CService
{
src/txdb-rocksdb.cpp
+123 -17
View File
@@ -31,6 +31,8 @@ namespace fs = std::filesystem;
// Global pointer for the RocksDB instance, shared across CRocksTxDB instances
// the same way the LevelDB backend shares its txdb singleton.
static rocksdb::DB* g_rocksdb = nullptr;
static rocksdb::ColumnFamilyHandle* g_cf_handles[5] = {}; // indexed by CF_ enum
static bool g_cf_enabled = false;
// Non-batched writes bypass WAL fsync. The TxnCommit path handles durability;
// crash recovery replays from block files anyway. Default WriteOptions may
@@ -95,6 +97,28 @@ static rocksdb::Options GetRocksOptions()
return opts;
}
// ─── Column family names ───────────────────────────────────────────────────
static const std::string CF_NAMES[] = {
rocksdb::kDefaultColumnFamilyName, // CF_DEFAULT (index 0)
"blockindex", // CF_BLOCKINDEX (index 1)
"txindex", // CF_TXINDEX (index 2)
"utxo", // CF_UTXO (index 3)
"addrindex", // CF_ADDRINDEX (index 4)
};
static constexpr int CF_COUNT = 5;
// Prefix-to-CF routing table. Keys starting with these prefixes go to
// the indicated CF index. Everything else stays in CF_DEFAULT (metadata).
struct CfPrefixEntry { const char* prefix; int len; int cf_index; };
static CfPrefixEntry prefixMap_[] = {
{"b", 1, 1}, // CF_BLOCKINDEX
{"t", 1, 2}, // CF_TXINDEX
{"u", 1, 3}, // CF_UTXO
{"addrbal", 7, 4}, // CF_ADDRINDEX
{"addrutxo", 8, 4}, // CF_ADDRINDEX
{"addrtxid", 8, 4}, // CF_ADDRINDEX
};
static void open_rocksdb(rocksdb::Options& options, bool fRemoveOld = false)
{
fs::path directory = GetDataDir() / "rocksdb";
@@ -105,11 +129,59 @@ static void open_rocksdb(rocksdb::Options& options, bool fRemoveOld = false)
fs::create_directory(directory);
printf("Opening RocksDB in %s\n", directory.string().c_str());
rocksdb::Status status = OpenRocksDB(options, directory.string(), &g_rocksdb);
if (!status.ok()) {
throw runtime_error(strprintf("open_rocksdb(): error opening database: %s",
status.ToString().c_str()));
// Try opening with column families. First, list existing CFs.
std::vector<std::string> existingCFs;
rocksdb::Options listOpts = options;
listOpts.create_if_missing = false;
rocksdb::DB::ListColumnFamilies(listOpts, directory.string(), &existingCFs);
bool needsCreate = (existingCFs.size() <= 1); // Only "default" or empty
std::vector<rocksdb::ColumnFamilyDescriptor> cfDescs;
for (int i = 0; i < CF_COUNT; i++) {
// Include this CF if it already exists OR if we're creating new
bool exists = false;
for (auto& name : existingCFs)
if (name == CF_NAMES[i]) { exists = true; break; }
if (exists || needsCreate) {
rocksdb::ColumnFamilyOptions cfOpts = options;
// Per-CF tuning:
if (i == 3) { // UTXO: optimize for point lookups
cfOpts.OptimizeForPointLookup(static_cast<size_t>(GetArg("-dbcache", 2048)));
} else if (i == 4) { // addrindex: optimize for scans
cfOpts.OptimizeLevelStyleCompaction(cfOpts.write_buffer_size);
}
cfDescs.push_back(rocksdb::ColumnFamilyDescriptor(CF_NAMES[i], cfOpts));
}
}
std::vector<rocksdb::ColumnFamilyHandle*> handles;
rocksdb::Status status = rocksdb::DB::Open(options, directory.string(),
cfDescs, &handles, &g_rocksdb);
if (!status.ok()) {
// Fallback: open without CFs (old-style single-CF database)
printf("RocksDB CF open failed (%s), falling back to single-CF\n", status.ToString().c_str());
status = OpenRocksDB(options, directory.string(), &g_rocksdb);
if (!status.ok()) {
throw runtime_error(strprintf("open_rocksdb(): error opening database: %s",
status.ToString().c_str()));
}
return;
}
// Store handles in the global array (CF names map directly to indices)
for (size_t i = 0; i < handles.size() && i < CF_COUNT; i++) {
// Match handle to our index by name
std::string hname = handles[i]->GetName();
for (int j = 0; j < CF_COUNT; j++) {
if (hname == CF_NAMES[j]) {
g_cf_handles[j] = handles[i];
break;
}
}
}
g_cf_enabled = true;
}
CRocksTxDB::CRocksTxDB(const char* pszMode)
@@ -245,6 +317,18 @@ bool CRocksTxDB::ScanBatch(const std::string& key, std::string* value, bool* del
return true;
}
// ─── CF routing helper ──────────────────────────────────────────────────────
rocksdb::ColumnFamilyHandle* CRocksTxDB::GetCF(const std::string& key) const
{
if (!g_cf_enabled)
return nullptr; // nullptr = default CF
for (auto& entry : prefixMap_) {
if ((int)key.size() >= entry.len && key.compare(0, entry.len, entry.prefix) == 0)
return g_cf_handles[entry.cf_index];
}
return nullptr; // default CF for metadata keys
}
bool CRocksTxDB::ReadRaw(const std::string& key, std::string& value) const
{
bool readFromDb = true;
@@ -255,10 +339,21 @@ bool CRocksTxDB::ReadRaw(const std::string& key, std::string& value) const
return false;
}
if (readFromDb) {
rocksdb::Status status = pdb->Get(rocksdb::ReadOptions(), key, &value);
rocksdb::ReadOptions ro;
auto* cf = GetCF(key);
rocksdb::Status status = cf ? pdb->Get(ro, cf, key, &value)
: pdb->Get(ro, key, &value);
if (!status.ok()) {
if (status.IsNotFound())
if (status.IsNotFound()) {
// If CFs are enabled and key wasn't in the target CF, also
// check the default CF (handles data written before CF migration)
if (g_cf_enabled && cf) {
rocksdb::Status status2 = pdb->Get(ro, key, &value);
if (!status2.ok()) return false;
return true;
}
return false;
}
printf("RocksDB read failure: %s\n", status.ToString().c_str());
return false;
}
@@ -268,12 +363,17 @@ bool CRocksTxDB::ReadRaw(const std::string& key, std::string& value) const
bool CRocksTxDB::WriteRaw(const std::string& key, const std::string& value)
{
auto* cf = GetCF(key);
if (activeBatch) {
activeBatch->Put(key, value);
if (cf)
activeBatch->Put(cf, key, value);
else
activeBatch->Put(key, value);
pendingBatch[key] = value;
return true;
}
rocksdb::Status status = pdb->Put(g_fastWriteOpts, key, value);
rocksdb::Status status = cf ? pdb->Put(g_fastWriteOpts, cf, key, value)
: pdb->Put(g_fastWriteOpts, key, value);
if (!status.ok()) {
printf("RocksDB write failure: %s\n", status.ToString().c_str());
return false;
@@ -285,12 +385,17 @@ bool CRocksTxDB::EraseRaw(const std::string& key)
{
if (!pdb)
return false;
auto* cf = GetCF(key);
if (activeBatch) {
activeBatch->Delete(key);
if (cf)
activeBatch->Delete(cf, key);
else
activeBatch->Delete(key);
pendingBatch[key] = std::nullopt;
return true;
}
rocksdb::Status status = pdb->Delete(rocksdb::WriteOptions(), key);
rocksdb::Status status = cf ? pdb->Delete(rocksdb::WriteOptions(), cf, key)
: pdb->Delete(rocksdb::WriteOptions(), key);
return (status.ok() || status.IsNotFound());
}
@@ -302,17 +407,18 @@ bool CRocksTxDB::ExistsRaw(const std::string& key) const
bool deleted = false;
bool inBatch = ScanBatch(key, &unused, &deleted);
if (inBatch) {
// Key is in the pending batch — present iff not marked deleted.
// Critically, a delete marker must shadow the underlying DB's
// version of the key (otherwise reads inside an open batch would
// still see the stale pre-erase value, defeating the whole point
// of the batch). Mirror ReadRaw's deleted==true → return false.
return !deleted;
}
// Not in the pending batch — fall through to underlying DB.
}
rocksdb::Status status = pdb->Get(rocksdb::ReadOptions(), key, &unused);
auto* cf = GetCF(key);
rocksdb::ReadOptions ro;
rocksdb::Status status = cf ? pdb->Get(ro, cf, key, &unused)
: pdb->Get(ro, key, &unused);
if (status.IsNotFound() && g_cf_enabled && cf) {
// Fallback to default CF for pre-migration data
status = pdb->Get(ro, key, &unused);
}
return status.IsNotFound() == false;
}
src/txdb-rocksdb.h
+15
View File
@@ -11,10 +11,12 @@
#include <optional>
#include <string>
#include <unordered_map>
#include <vector>
#include <rocksdb/db.h>
#include <rocksdb/options.h>
#include <rocksdb/write_batch.h>
#include <rocksdb/utilities/db_ttl.h>
// RocksDB backend for the chain database.
//
@@ -72,6 +74,19 @@ private:
rocksdb::Options options;
int nVersion;
// ─── Column family support ──────────────────────────────────────────────
// Data is split into CFs for independent compaction and caching.
// cf_handles[0] is always the default CF (for backward compatibility
// with pre-CF databases that have all data in "default").
enum CfId : int { CF_DEFAULT = 0, CF_BLOCKINDEX, CF_TXINDEX, CF_UTXO, CF_ADDRINDEX, CF_COUNT };
rocksdb::ColumnFamilyHandle* cf_handles[CF_COUNT] = {};
bool cf_enabled = false; // True if CFs were created/opened successfully
// Route a key to the correct column family handle based on its prefix.
// Falls back to CF_DEFAULT for keys that don't match any known prefix
// (metadata like "version", "hashBestChain", etc.) or if CFs aren't enabled.
rocksdb::ColumnFamilyHandle* GetCF(const std::string& key) const;
// Parallel record of every pending write (value) or delete (nullopt) on
// activeBatch. Used by ScanBatch to answer "is this key already in the
// active batch?" without iterating the WriteBatch via Handler — Ubuntu's