afd59cb · hyperswarm-peer

Changed files (27) +187 -19

docs/PARITY-hyperswarm.md +44 -0

…	…	@@ -986,3 +986,47 @@ possible.
986	986
987	987	⚠ NOT STARTED. Recorded rather than begun because it is a four-part change
988	988	across two files' public signatures, and half of it landing is worse than none.
	989
	990	## 16. Slice 12 — B's liveness ping, built on real holders (DONE)
	991
	992	§15's prerequisite first, then the ping.
	993
	994	1. The commit callback now carries the responding node's ADDRESS.
	995	`dht_rpc`'s per-node commit fn was `(target, token, node_id)`; the reply's origin
	996	is the only place a holder's address appears, so it is now
	997	`(target, token, node_id, from)`.
	998
	999	2. `_note_announce_holder` replaces `_store_announced_nodes`. Holders are
	1000	recorded AS THEY COMMIT, from the reply, with the token they issued — instead of
	1001	being guessed after the fact from `table().closest()` with a zeroed token. A node
	1002	re-committing on a refresh updates its token rather than duplicating: the token
	1003	rotates and a stale one is worthless. This is JS's `_serverRelays[2].set(msg.from…)`
	1004	(`announcer.js:267`).
	1005
	1006	3. The ping round. `_ping_self_holders` pings every recorded holder every
	1007	`SELF_ANNOUNCE_PING_MS` (3000, = JS), tagged `HYPERDHT_TAG_SELF_PING` so replies
	1008	are distinguishable from every other PING.
	1009
	1010	4. The verdict. JS: `if (active < Math.min(pings.length, MIN_ACTIVE)) refresh()`.
	1011	Ours judges the round at the START of the next one — a slow reply inside the
	1012	window still counts — and on failure sets `_self_announce_at = now`, so the
	1013	re-announce fires in the SAME tick. Ordering matters: the ping runs before
	1014	`_tick_self_announce`, which is what lets it pull the announce forward.
	1015
	1016	That is 3 seconds to notice we are unreachable instead of 5 minutes.
	1017
	1018	Gate: `test_self_announce.cpp` extended to 10/10, including the degenerate
	1019	cases that would otherwise be a crash or a storm: no holders recorded means no
	1020	ping rounds, and an empty holder set must NOT trigger a re-announce (an empty
	1021	set trivially has "fewer than 3 alive").
	1022
	1023	⚠ What is NOT covered: the populated-holder path. These gates have no real
	1024	DHT, so nothing ever answers an announce and `self_holder_count()` is 0
	1025	throughout. The recording, the ping round and the verdict are exercised only in
	1026	their empty form. Proving the live path needs a testnet with real responders —
	1027	named here so nobody reads 10/10 as more than it is.
	1028
	1029	⏳ Still open: three-generation cycling, `refresh-chain`, A2, C's explicit-error
	1030	abort.
	1031
	1032	Suite: 13 binaries green.

src/dht_rpc.cpp +3 -2

…	…	@@ -236,7 +236,8 @@ uint32_t DhtRpc::query_with_commit_fn(
236	236	uint32_t traverse_command,
237	237	const uint8_t *traverse_value, size_t traverse_value_len,
238	238	uint32_t commit_command,
239		std::function<std::vector<uint8_t>(const uint8_t , const uint8_t , const uint8_t *)> commit_fn,
	239	std::function<std::vector<uint8_t>(const uint8_t , const uint8_t ,
	240	const uint8_t *, const IPv4Address &)> commit_fn,
240	241	uint32_t tag) {
241	242	if (_query_count >= 16) return 0;
242	243
…	…	@@ -1043,7 +1044,7 @@ void DhtRpc::_commit_query(DhtQuery *q) {
1043	1044	if (q->commit_value_fn) {
1044	1045	// Compute the node's ID from its address (matches JS peer.id()).
1045	1046	NodeId node_id = cenc::node_id_from_address(rp.address);
1046		per_node_val = q->commit_value_fn(q->target, rp.token, node_id.bytes);
	1047	per_node_val = q->commit_value_fn(q->target, rp.token, node_id.bytes, rp.address);
1047	1048	commit_val = per_node_val.data();
1048	1049	commit_val_len = per_node_val.size();
1049	1050	} else {

src/dht_rpc.h +9 -2

…	…	@@ -144,7 +144,8 @@ struct DhtQuery {
144	144	// Returns: the commit value for this specific node.
145	145	std::function<std::vector<uint8_t>(const uint8_t *target,
146	146	const uint8_t *token,
147		const uint8_t *node_id)> commit_value_fn;
	147	const uint8_t *node_id,
	148	const IPv4Address &from)> commit_value_fn;
148	149
149	150	DhtQueryPeer closest[DHT_QUERY_K * 3]; // Working set.
150	151	int closest_count;
…	…	@@ -208,7 +209,13 @@ public:
208	209	uint32_t traverse_command,
209	210	const uint8_t *traverse_value, size_t traverse_value_len,
210	211	uint32_t commit_command,
211		std::function<std::vector<uint8_t>(const uint8_t , const uint8_t , const uint8_t *)> commit_fn,
	212	// (target, token, node_id, from) -> the value to commit at THAT node.
	213	// ⛔ `from` was missing and that is why nothing could record which nodes
	214	// actually hold our record: the commit reply is the only place the
	215	// holder's address appears. _store_announced_nodes had to guess with
	216	// table().closest() and zero the token. See PARITY-hyperswarm.md §15.
	217	std::function<std::vector<uint8_t>(const uint8_t , const uint8_t ,
	218	const uint8_t *, const IPv4Address &)> commit_fn,
212	219	uint32_t tag = 0);
213	220
214	221	// Process an incoming UDP message. Called from socket recv callback.

src/hyperdht.cpp +85 -13

…	…	@@ -69,9 +69,9 @@ void HyperDht::init(DhtRpc *rpc, const Ed25519KeyPair &keypair) {
69	69	if (_query_infos[i].active && _query_infos[i].query_id == query_id) {
70	70	auto &qi = _query_infos[i];
71	71	// For announce queries, store the closest responded nodes.
72		if (qi.is_announce) {
73		_store_announced_nodes(qi.topic, query_id);
74		}
	72	// Holders are recorded per node in the commit callback now
	73	// (_note_announce_holder), not guessed from the routing table
	74	// after the fact.
75	75	// Handle storage query completions.
76	76	if (qi.is_immutable_put) {
77	77	if (on_immutable_put_complete) {
…	…	@@ -204,7 +204,12 @@ uint32_t HyperDht::announce(const uint8_t *topic) {
204	204	HYPERDHT_CMD_LOOKUP, nullptr, 0, // Traversal: LOOKUP, no value
205	205	HYPERDHT_CMD_ANNOUNCE, // Commit: ANNOUNCE
206	206	[this, topic_copy](const uint8_t target, const uint8_t token,
207		const uint8_t *node_id) -> std::vector<uint8_t> {
	207	const uint8_t *node_id,
	208	const IPv4Address &from) -> std::vector<uint8_t> {
	209	// ⛔ THIS is where we learn who holds our record. The reply's `from`
	210	// is the node that took the announce; nothing else knows it. See
	211	// _note_announce_holder and PARITY-hyperswarm.md §15.
	212	_note_announce_holder(topic_copy, from, token);
208	213	return _encode_announce_value_for_node(topic_copy, token, node_id);
209	214	},
210	215	0);
…	…	@@ -259,7 +264,8 @@ void HyperDht::unannounce(const uint8_t *topic) {
259	264	HYPERDHT_CMD_LOOKUP, nullptr, 0, // traversal: LOOKUP, no value
260	265	HYPERDHT_CMD_UNANNOUNCE, // commit: UNANNOUNCE
261	266	[this, topic_copy](const uint8_t target, const uint8_t token,
262		const uint8_t *node_id) -> std::vector<uint8_t> {
	267	const uint8_t *node_id,
	268	const IPv4Address &) -> std::vector<uint8_t> {
263	269	return _encode_ann_value_for_node(_ns_unannounce, topic_copy, token, node_id);
264	270	},
265	271	0);
…	…	@@ -365,6 +371,42 @@ void HyperDht::listen() {
365	371	// JS keeps this alive in a dedicated component, hyperdht's Announcer, one
366	372	// per server, targeting hash(keyPair.publicKey) — see SELF_ANNOUNCE_UPDATE_MS.
367	373	_self_announce_at = _last_tick_ms + SELF_ANNOUNCE_UPDATE_MS;
	374	_self_ping_at = _last_tick_ms + SELF_ANNOUNCE_PING_MS;
	375	}
	376
	377	int HyperDht::self_holder_count() const {
	378	auto it = _announced_topics.find(_topic_hex(_self_target));
	379	return it == _announced_topics.end() ? 0 : it->second.node_count;
	380	}
	381
	382	void HyperDht::_ping_self_holders(uint64_t now_ms) {
	383	auto it = _announced_topics.find(_topic_hex(_self_target));
	384	if (it == _announced_topics.end() \|\| it->second.node_count == 0) return;
	385
	386	const AnnouncedTopic &at = it->second;
	387
	388	// Judge the round that has just elapsed before starting the next one.
	389	// JS: if (active < Math.min(pings.length, MIN_ACTIVE)) this.refresh()
	390	if (_self_ping_sent > 0) {
	391	const int needed = _self_ping_sent < SELF_ANNOUNCE_MIN_ACTIVE
	392	? _self_ping_sent
	393	: SELF_ANNOUNCE_MIN_ACTIVE;
	394	if (_self_ping_alive < needed) {
	395	DHT_DISC("[HyperDht] only %d/%d holders answered — re-announcing now "
	396	"instead of in %llums\n", _self_ping_alive, _self_ping_sent,
	397	(unsigned long long)(_self_announce_at - now_ms));
	398	_self_announce_at = now_ms; // _tick_self_announce fires this tick
	399	}
	400	}
	401
	402	_self_ping_sent = 0;
	403	_self_ping_alive = 0;
	404	for (int i = 0; i < at.node_count; i++) {
	405	_rpc->ping(at.nodes[i].address, HYPERDHT_TAG_SELF_PING);
	406	_self_ping_sent++;
	407	}
	408	_self_ping_rounds++;
	409	_self_ping_at = now_ms + SELF_ANNOUNCE_PING_MS;
368	410	}
369	411
370	412	void HyperDht::_tick_self_announce(uint64_t now_ms) {
…	…	@@ -393,6 +435,12 @@ void HyperDht::tick(uint64_t now_ms) {
393	435
394	436	_last_tick_ms = now_ms;
395	437
	438	// Are the nodes holding our record still there? Runs BEFORE the announce
	439	// tick so a failed round can pull the re-announce forward into this tick.
	440	if (_listening && _self_ping_at != 0 && now_ms >= _self_ping_at) {
	441	_ping_self_holders(now_ms);
	442	}
	443
396	444	// Keep our own record alive. Without this a listening peer is announced
397	445	// exactly once and becomes unreachable when that record expires.
398	446	_tick_self_announce(now_ms);
…	…	@@ -408,6 +456,14 @@ void HyperDht::tick(uint64_t now_ms) {
408	456	// ---- RPC response handler ----
409	457
410	458	void HyperDht::_on_rpc_response(const DhtResponse &res, uint32_t tag) {
	459	// A holder answered its liveness ping. Counted, not acted on here: the
	460	// verdict is passed at the START of the next round, so a slow reply inside
	461	// the window still counts. See _ping_self_holders.
	462	if (tag == HYPERDHT_TAG_SELF_PING) {
	463	_self_ping_alive++;
	464	return;
	465	}
	466
411	467	// Check if this is a response to a handshake.
412	468	DhtHandshake *hs = _find_handshake(tag);
413	469	if (hs && hs->is_initiator) {
…	…	@@ -887,20 +943,36 @@ void HyperDht::_sign_announce(const uint8_t *signable, size_t signable_len,
887	943	_keypair.secret_key);
888	944	}
889	945
890		void HyperDht::_store_announced_nodes(const uint8_t *topic, uint32_t query_id) {
891		// Store the closest nodes from the routing table for later unannounce.
	946	void HyperDht::_note_announce_holder(const uint8_t *topic, const IPv4Address &from,
	947	const uint8_t *token) {
	948	// ⛔ THE NODES THAT ACTUALLY TOOK THE ANNOUNCE, recorded as they commit.
	949	//
	950	// This used to be _store_announced_nodes(), which ran after the query and
	951	// filled the table from `table().closest()` — our own routing table's idea
	952	// of who is nearby, not who accepted anything — and then wrote a zeroed
	953	// token with the comment "Token not available here". So we could not say
	954	// which nodes held our record, could not check whether they were alive, and
	955	// could not produce a per-node token for anything later.
	956	//
	957	// JS records the committing node from the reply itself
	958	// (announcer.js:265-267). `from` here is that same reply's origin.
892	959	std::string key = _topic_hex(topic);
893	960	AnnouncedTopic &at = _announced_topics[key];
894	961	memcpy(at.topic, topic, 32);
895	962
896		// Get closest nodes from the routing table.
897		KademliaNode nodes[20];
898		int n = _rpc->table().closest(topic, nodes, 20);
899		at.node_count = (n > 20) ? 20 : n;
	963	// A node re-committing on a refresh updates its token rather than adding a
	964	// duplicate — the token rotates and the stale one is worthless.
900	965	for (int i = 0; i < at.node_count; i++) {
901		at.nodes[i].address = nodes[i].address;
902		memset(at.nodes[i].token, 0, 32); // Token not available here.
	966	if (memcmp(at.nodes[i].address.ip, from.ip, 4) == 0 &&
	967	at.nodes[i].address.port == from.port) {
	968	memcpy(at.nodes[i].token, token, 32);
	969	return;
	970	}
903	971	}
	972	if (at.node_count >= 20) return;
	973	at.nodes[at.node_count].address = from;
	974	memcpy(at.nodes[at.node_count].token, token, 32);
	975	at.node_count++;
904	976	}
905	977
906	978	std::string HyperDht::_topic_hex(const uint8_t *topic) {

src/hyperdht.h +23 -2

…	…	@@ -36,6 +36,17 @@ static constexpr int HYPERDHT_MAX_PEERS = 64;
36	36	// 100 iterations of a 3s sleep, so a full re-announce every ~5 minutes.
37	37	static constexpr uint64_t SELF_ANNOUNCE_UPDATE_MS = 300000; // 5 min, = JS 100 x 3s.
38	38
	39	// ⛔ AND CHECK THE HOLDERS ARE STILL THERE. Re-announcing on a timer alone means
	40	// losing every node that holds our record costs a full interval before anyone
	41	// notices. JS pings each holder every 3s and re-announces the moment fewer than
	42	// MIN_ACTIVE answer (announcer.js:104-125) — 3 seconds versus 5 minutes to
	43	// notice we have become unreachable.
	44	static constexpr uint64_t SELF_ANNOUNCE_PING_MS = 3000;
	45	// Tag carried on holder pings so their replies can be told apart from every
	46	// other PING this node makes.
	47	static constexpr uint32_t HYPERDHT_TAG_SELF_PING = 0x5E1F0001;
	48	static constexpr int SELF_ANNOUNCE_MIN_ACTIVE = 3; // JS MIN_ACTIVE
	49
39	50	static constexpr uint64_t TOPIC_REFRESH_INTERVAL_MS = 600000; // 10 minutes.
40	51	static constexpr uint64_t TOPIC_REFRESH_JITTER_MS = 120000; // 0–2 minutes random.
41	52	static constexpr uint64_t TOPIC_REFRESH_GRACE_MS = 30000; // 30s grace (laptop wake).
…	…	@@ -176,6 +187,8 @@ public:
176	187	// expires, which is why this is observable at all.
177	188	uint64_t self_announce_due_at() const { return _self_announce_at; }
178	189	uint64_t self_announce_count() const { return _self_announce_count; }
	190	uint64_t self_ping_rounds() const { return _self_ping_rounds; }
	191	int self_holder_count() const;
179	192
180	193	// Process time-based events.
181	194	void tick(uint64_t now_ms);
…	…	@@ -323,8 +336,11 @@ private:
323	336	// Sign an announce record.
324	337	void _sign_announce(const uint8_t signable, size_t signable_len, uint8_t signature);
325	338
326		// Store closest nodes after announce for unannounce.
327		void _store_announced_nodes(const uint8_t *topic, uint32_t query_id);
	339	// Record a node that ACCEPTED our announce, with the token it issued.
	340	// Called from the commit callback, which is the only place the holder's
	341	// address exists. See PARITY-hyperswarm.md §15.
	342	void _note_announce_holder(const uint8_t *topic, const IPv4Address &from,
	343	const uint8_t *token);
328	344
329	345	// Handshake management.
330	346	DhtHandshake _alloc_handshake(bool initiator, const uint8_t remote_pk,
…	…	@@ -348,7 +364,12 @@ private:
348	364	uint8_t _self_target[32] = { 0 };
349	365	uint64_t _self_announce_at = 0; // 0 = not listening
350	366	uint64_t _self_announce_count = 0;
	367	uint64_t _self_ping_at = 0;
	368	uint64_t _self_ping_rounds = 0;
	369	int _self_ping_sent = 0; // holders pinged in the round now in flight
	370	int _self_ping_alive = 0; // how many answered
351	371	void _tick_self_announce(uint64_t now_ms);
	372	void _ping_self_holders(uint64_t now_ms);
352	373	void _finalize_handshake(DhtHandshake *hs);
353	374
354	375	// Respond to any non-internal query with closer nodes + token + optional value.

test/test_ann_signable +0 -0