From fb07d502358b25ad52b58859c289d70f948a2a5e Mon Sep 17 00:00:00 2001 From: Krystie Date: Sat, 27 Jun 2026 19:19:30 -0700 Subject: [PATCH] feat: compact blocks, column families, fork detector, cross-network discovery, SAM v3, configurable peers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- src/i2p/i2p_embedded.cpp | 379 ++++++++++++++++++++++++++++-- src/i2p/i2p_embedded.h | 87 +++++++ src/init.cpp | 33 +++ src/main.cpp | 492 +++++++++++++++++++++++++++++---------- src/miner.cpp | 10 +- src/net.cpp | 158 ++++++++++++- src/net.h | 2 + src/protocol.h | 12 + src/txdb-rocksdb.cpp | 140 +++++++++-- src/txdb-rocksdb.h | 15 ++ 10 files changed, 1165 insertions(+), 163 deletions(-) diff --git a/src/i2p/i2p_embedded.cpp b/src/i2p/i2p_embedded.cpp index 211cc4d..8ff3f13 100644 --- a/src/i2p/i2p_embedded.cpp +++ b/src/i2p/i2p_embedded.cpp @@ -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 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= DESTINATION=TRANSIENT + // S → C: SESSION STATUS RESULT=OK DESTINATION= + // ---------------------------------------------------------------- + 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= DESTINATION=.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) { diff --git a/src/i2p/i2p_embedded.h b/src/i2p/i2p_embedded.h index d2e9cb1..6e0d0b1 100644 --- a/src/i2p/i2p_embedded.h +++ b/src/i2p/i2p_embedded.h @@ -8,6 +8,80 @@ #include #include +// 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 + 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) diff --git a/src/init.cpp b/src/init.cpp index 372cede..6f0cb32 100644 --- a/src/init.cpp +++ b/src/init.cpp @@ -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= " + _("Listen for connections on (default: 24112 or testnet: 24111)") + "\n" + " -maxconnections= " + _("Maintain at most connections to peers (default: 125)") + "\n" + + " -maxoutboundconnections= " + _("Maximum outbound connections (default: 8, range 4-32)") + "\n" + " -addnode= " + _("Add a node to connect to and attempt to keep the connection open") + "\n" + " -connect= " + _("Connect only to the specified node(s)") + "\n" + " -seednode= " + _("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 diff --git a/src/main.cpp b/src/main.cpp index 1f740ba..489b72a 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -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 cPeerBlockCounts(5, 0); // Amount of blocks that other nodes claim to have CScriptVerifyCache scriptVerifyCache; @@ -103,6 +158,278 @@ static std::map 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 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 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. { diff --git a/src/miner.cpp b/src/miner.cpp index f6e4dc9..1137da8 100644 --- a/src/miner.cpp +++ b/src/miner.cpp @@ -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 { diff --git a/src/net.cpp b/src/net.cpp index 307b973..6eaf862 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -13,6 +13,7 @@ #include "onionseed.h" #include "tor/onion_v3.h" #include "snapshotnet.h" +#include "i2p/i2pseed.h" #include #include @@ -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& 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)); + } +} diff --git a/src/net.h b/src/net.h index ef46075..8c135df 100644 --- a/src/net.h +++ b/src/net.h @@ -21,7 +21,9 @@ class CNode; class CBlockIndex; bool IsInitialBlockDownload(); +void ThreadForkDetector(void*); extern int nBestHeight; +extern int nForkAlertCount; diff --git a/src/protocol.h b/src/protocol.h index 678e18c..c419d64 100644 --- a/src/protocol.h +++ b/src/protocol.h @@ -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 { diff --git a/src/txdb-rocksdb.cpp b/src/txdb-rocksdb.cpp index 986ad73..660936c 100644 --- a/src/txdb-rocksdb.cpp +++ b/src/txdb-rocksdb.cpp @@ -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 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 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(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 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; } diff --git a/src/txdb-rocksdb.h b/src/txdb-rocksdb.h index 8e2a303..f225830 100644 --- a/src/txdb-rocksdb.h +++ b/src/txdb-rocksdb.h @@ -11,10 +11,12 @@ #include #include #include +#include #include #include #include +#include // 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