54eb4c5 · hyperswarm-peer

Changed files (5) +829 -1

swarmd/.gitignore +1 -0

…	…	@@ -2,5 +2,6 @@ node_modules/
2	2	test/test_uds_backpressure
3	3	test/test_control
4	4	test/test_send_truth
	5	test/test_client
5	6	swarmd
6	7	tools/swarmd-probe

swarmd/build.sh +10 -1

…	…	@@ -7,7 +7,7 @@ cd "$(dirname "$0")"
7	7
8	8	KB=/storage/projects/code/hypercore-kb/holepunchto/repos
9	9	PEER=..
10		INC="-Isrc -Ischema -I$PEER/src -I$PEER/liboverlay/include -I$PEER/thirdparty/libudx/include -I$KB/tier5/libcompact/include -I$KB/tier4/libutf/include -I$KB/tier4/librpc/include"
	10	INC="-Isrc -Ischema -Iclient -I$PEER/src -I$PEER/liboverlay/include -I$PEER/thirdparty/libudx/include -I$KB/tier5/libcompact/include -I$KB/tier4/libutf/include -I$KB/tier4/librpc/include"
11	11	OBJ=${OBJ:-/tmp/swarmd-obj}
12	12	mkdir -p "$OBJ"
13	13
…	…	@@ -37,6 +37,9 @@ echo "Building generated wire..."
37	37	gcc -std=c99 -O2 $INC -c -o "$OBJ/swarmd_schema.o" schema/swarmd_schema.c
38	38	gcc -std=c99 -O2 $INC -c -o "$OBJ/swarmd_hrpc.o" schema/swarmd_hrpc.c
39	39
	40	echo "Building the client library..."
	41	g++ -std=c++17 -O2 $INC -c -o "$OBJ/client.o" client/client.cpp
	42
40	43	echo "Building daemon sources..."
41	44	g++ -std=c++17 -O2 $INC -c -o "$OBJ/uds_server.o" src/uds_server.cpp
42	45	g++ -std=c++17 -O2 $INC -c -o "$OBJ/control.o" src/control.cpp
…	…	@@ -96,6 +99,12 @@ g++ -std=c++17 -O2 $INC -o test/test_send_truth \
96	99	"$OBJ/swarmd_schema.o" "$OBJ/swarmd_hrpc.o" $CLIBS $TLIBS \
97	100	-lsodium -luv -lpthread
98	101
	102	g++ -std=c++17 -O2 $INC -o test/test_client \
	103	test/test_client.cpp "$OBJ/client.o" "$OBJ/daemon.o" "$OBJ/uds_server.o" \
	104	"$OBJ/control.o" "$OBJ/content.o" \
	105	"$OBJ/swarmd_schema.o" "$OBJ/swarmd_hrpc.o" $CLIBS $TLIBS \
	106	-lsodium -luv -lpthread
	107
99	108	echo "Building tools..."
100	109	g++ -std=c++17 -O2 $INC -o tools/swarmd-probe tools/swarmd_probe.cpp \
101	110	"$OBJ/swarmd_schema.o" "$OBJ/swarmd_hrpc.o" $CLIBS

swarmd/client/client.cpp +461 -0

…	…	@@ -0,0 +1,461 @@
	1	#include "client.h"
	2
	3	#include <cerrno>
	4	#include <cstdio>
	5	#include <cstdlib>
	6	#include <cstring>
	7
	8	#include <poll.h>
	9	#include <sys/socket.h>
	10	#include <sys/un.h>
	11	#include <unistd.h>
	12
	13	extern "C" {
	14	#include <compact.h>
	15	#include <rpc.h>
	16	#include "swarmd_hrpc.h"
	17	#include "swarmd_schema.h"
	18	}
	19
	20	#include "../src/uds_server.h" // UDS_MAX_FRAME — one definition of the cap
	21
	22	namespace swarmd {
	23
	24	// ---- framing -------------------------------------------------------------
	25
	26	static uint32_t le32(const uint8_t *p) {
	27	return (uint32_t)p[0] \| ((uint32_t)p[1] << 8) \| ((uint32_t)p[2] << 16) \| ((uint32_t)p[3] << 24);
	28	}
	29
	30	static bool write_all(int fd, const uint8_t *p, size_t n) {
	31	size_t off = 0;
	32	while (off < n) {
	33	const ssize_t w = ::write(fd, p + off, n - off);
	34	if (w > 0) { off += (size_t)w; continue; }
	35	if (w < 0 && (errno == EINTR)) continue;
	36	if (w < 0 && (errno == EAGAIN \|\| errno == EWOULDBLOCK)) {
	37	// The socket is blocking, so this is not expected — but a short wait
	38	// is still the right answer to a full kernel buffer, and dropping
	39	// half a frame is not.
	40	struct pollfd pfd = { fd, POLLOUT, 0 };
	41	if (::poll(&pfd, 1, 1000) > 0) continue;
	42	}
	43	return false;
	44	}
	45	return true;
	46	}
	47
	48	Client::~Client() { close(); }
	49
	50	void Client::close() {
	51	if (_fd >= 0) ::close(_fd);
	52	_fd = -1;
	53	_in.clear();
	54	_reply.clear();
	55	}
	56
	57	void Client::_fail(const std::string &why) {
	58	if (_fd < 0) return; // already reported
	59	_err = why;
	60	close();
	61	if (on.on_disconnect) on.on_disconnect(why);
	62	}
	63
	64	bool Client::_send_frame(uint8_t *buf, size_t len) {
	65	if (_fd < 0 \|\| buf == nullptr) { free(buf); return false; }
	66	uint8_t hdr[4] = { (uint8_t)(len & 0xff), (uint8_t)((len >> 8) & 0xff),
	67	(uint8_t)((len >> 16) & 0xff), (uint8_t)((len >> 24) & 0xff) };
	68	const bool ok = write_all(_fd, hdr, 4) && write_all(_fd, buf, len);
	69	free(buf); // ⛔ the caller frees what _encode_ allocated
	70	if (!ok) _fail("the daemon closed while we were writing to it");
	71	return ok;
	72	}
	73
	74	// ---- reading -------------------------------------------------------------
	75
	76	bool Client::_read_some(int timeout_ms) {
	77	if (_fd < 0) return false;
	78	struct pollfd pfd = { _fd, POLLIN, 0 };
	79	const int r = ::poll(&pfd, 1, timeout_ms);
	80	if (r < 0) {
	81	if (errno == EINTR) return true;
	82	_fail("poll failed on the daemon socket");
	83	return false;
	84	}
	85	if (r == 0) return true; // nothing yet, not an error
	86
	87	uint8_t tmp[65536];
	88	const ssize_t n = ::read(_fd, tmp, sizeof(tmp));
	89	if (n == 0) { _fail("the daemon closed the connection"); return false; }
	90	if (n < 0) {
	91	if (errno == EINTR \|\| errno == EAGAIN \|\| errno == EWOULDBLOCK) return true;
	92	_fail("read failed on the daemon socket");
	93	return false;
	94	}
	95	_in.insert(_in.end(), tmp, tmp + n);
	96	return true;
	97	}
	98
	99	void Client::_dispatch_event(const void *vmsg) {
	100	const rpc_message_t msg = (const rpc_message_t )vmsg;
	101	swarmd_hrpc_handlers_t h;
	102	memset(&h, 0, sizeof(h));
	103	h.ctx = this;
	104	h.on_on_message = [](void ctx, const swarmd_message_t m) {
	105	Client c = (Client )ctx;
	106	if (c->on.on_message)
	107	c->on.on_message(m->channel, m->slot, m->origin_kind, m->origin_id,
	108	m->bytes, m->bytes_len);
	109	};
	110	h.on_on_peer_up = [](void ctx, const swarmd_peer_up_t m) {
	111	Client c = (Client )ctx;
	112	if (c->on.on_peer_up) c->on.on_peer_up(m->peer, m->pubkey);
	113	};
	114	h.on_on_peer_down = [](void ctx, const swarmd_peer_down_t m) {
	115	Client c = (Client )ctx;
	116	if (c->on.on_peer_down) c->on.on_peer_down(m->peer);
	117	};
	118	h.on_on_drain = [](void ctx, const swarmd_channel_ref_t m) {
	119	Client c = (Client )ctx;
	120	if (c->on.on_drain) c->on.on_drain(m->channel);
	121	};
	122	h.on_on_content_progress = [](void ctx, const swarmd_content_progress_t m) {
	123	Client c = (Client )ctx;
	124	if (c->on.on_content_progress) c->on.on_content_progress(m->hash, m->received, m->total);
	125	};
	126	h.on_on_content_done = [](void ctx, const swarmd_content_done_t m) {
	127	Client c = (Client )ctx;
	128	if (c->on.on_content_done)
	129	c->on.on_content_done(m->hash, std::string((const char *)m->path.data, m->path.len));
	130	};
	131	h.on_on_content_failed = [](void ctx, const swarmd_content_failed_t m) {
	132	Client c = (Client )ctx;
	133	if (c->on.on_content_failed)
	134	c->on.on_content_failed(m->hash, std::string((const char *)m->reason.data, m->reason.len));
	135	};
	136	// An event carries no reply, so anything the dispatcher hands back is
	137	// discarded rather than sent — but it must still be freed.
	138	uint8_t *rep = nullptr;
	139	size_t rl = 0;
	140	swarmd_hrpc_dispatch(&h, msg, &rep, &rl);
	141	if (rep) free(rep);
	142	}
	143
	144	bool Client::_drain_frames(uint64_t await_id, const uint8_t *body, size_t body_len) {
	145	for (;;) {
	146	if (_in.size() < 4) return false;
	147	const uint32_t n = le32(_in.data());
	148	if (n > UDS_MAX_FRAME) {
	149	// A length this large is a framing bug, not a frame — and the stream
	150	// is desynchronised, so there is nothing to resume from.
	151	_fail("the daemon sent a frame length past the cap; the stream is out of sync");
	152	return false;
	153	}
	154	if (_in.size() < 4 + (size_t)n) return false;
	155
	156	// Copy out before consuming: a decoded view points INTO this buffer, and
	157	// the buffer moves as soon as the next frame is taken off it.
	158	std::vector<uint8_t> frame(_in.begin() + 4, _in.begin() + 4 + n);
	159	_in.erase(_in.begin(), _in.begin() + 4 + n);
	160
	161	compact_state_t st = { 0, frame.size(), frame.data() };
	162	rpc_message_t msg;
	163	if (rpc_decode_message(&st, &msg) < 0) {
	164	_fail("the daemon sent a message this client could not decode");
	165	return false;
	166	}
	167
	168	if (msg.type == rpc_response) {
	169	if (msg.id != await_id) continue; // a reply nobody is waiting for
	170	if (msg.error) {
	171	_err = "the daemon answered with an error";
	172	// Not a disconnect: the call failed, the session is fine.
	173	_reply.clear();
	174	if (body) *body = nullptr;
	175	if (body_len) *body_len = 0;
	176	return true;
	177	}
	178	_reply.assign(msg.data, msg.data + msg.len);
	179	if (body) *body = _reply.empty() ? nullptr : _reply.data();
	180	if (body_len) *body_len = _reply.size();
	181	return true;
	182	}
	183
	184	// ⛔ EVENTS ARE DELIVERED EVEN WHILE A CALL IS WAITING. A peer arriving,
	185	// or a fetch finishing, does not queue politely behind the answer to
	186	// channel_open.
	187	if (msg.type == rpc_request) _dispatch_event(&msg);
	188	}
	189	}
	190
	191	bool Client::_await(uint64_t id, int timeout_ms, const uint8_t *body, size_t body_len) {
	192	if (body) *body = nullptr;
	193	if (body_len) *body_len = 0;
	194	// Anything already buffered may hold the answer.
	195	if (_drain_frames(id, body, body_len)) return true;
	196	if (_fd < 0) return false;
	197
	198	int left = timeout_ms;
	199	while (left > 0) {
	200	const int slice = left > 200 ? 200 : left;
	201	if (!_read_some(slice)) return false;
	202	if (_drain_frames(id, body, body_len)) return true;
	203	if (_fd < 0) return false;
	204	left -= slice;
	205	}
	206	_err = "the daemon did not answer in time";
	207	return false;
	208	}
	209
	210	bool Client::poll(int timeout_ms) {
	211	if (_fd < 0) return false;
	212	if (!_read_some(timeout_ms)) return false;
	213	// id 0 is never a request id, so nothing matches and every frame is an event.
	214	_drain_frames(0, nullptr, nullptr);
	215	return _fd >= 0;
	216	}
	217
	218	// ---- calls ---------------------------------------------------------------
	219
	220	bool Client::connect(const std::string &socket_path, const std::string &name,
	221	const uint8_t app_identity[32], int timeout_ms) {
	222	close();
	223	_err.clear();
	224
	225	_fd = ::socket(AF_UNIX, SOCK_STREAM, 0);
	226	if (_fd < 0) { _err = "could not create a unix socket"; return false; }
	227	struct sockaddr_un a;
	228	memset(&a, 0, sizeof(a));
	229	a.sun_family = AF_UNIX;
	230	strncpy(a.sun_path, socket_path.c_str(), sizeof(a.sun_path) - 1);
	231	if (::connect(_fd, (struct sockaddr *)&a, sizeof(a)) != 0) {
	232	::close(_fd);
	233	_fd = -1;
	234	_err = "no daemon at " + socket_path;
	235	return false;
	236	}
	237
	238	swarmd_hello_request_t req;
	239	memset(&req, 0, sizeof(req));
	240	req.name.data = (utf8_t *)name.c_str();
	241	req.name.len = name.size();
	242	if (app_identity) memcpy(req.app_identity, app_identity, 32);
	243	req.version = 1;
	244
	245	const uint64_t id = _next_id++;
	246	uint8_t *buf = nullptr;
	247	size_t len = 0;
	248	if (swarmd_encode_hello(id, &req, &buf, &len) < 0) { _fail("could not encode HELLO"); return false; }
	249	if (!_send_frame(buf, len)) return false;
	250
	251	const uint8_t *body = nullptr;
	252	size_t body_len = 0;
	253	if (!_await(id, timeout_ms, &body, &body_len)) { close(); return false; }
	254
	255	swarmd_hello_response_t res;
	256	memset(&res, 0, sizeof(res));
	257	if (body) {
	258	compact_state_t in = { 0, body_len, (uint8_t *)body };
	259	if (swarmd_hello_response_decode(&in, &res) < 0) {
	260	_fail("the daemon's HELLO reply did not decode");
	261	return false;
	262	}
	263	}
	264	_client_id = res.client_id;
	265	memcpy(_daemon_pubkey, res.daemon_pubkey, 32);
	266	return true;
	267	}
	268
	269	// One shape for every call that answers with `ok`.
	270	#define SWARMD_OK_CALL(ENCODE, REQPTR) \
	271	do { \
	272	if (_fd < 0) { _err = "not connected"; return false; } \
	273	const uint64_t id = _next_id++; \
	274	uint8_t *buf = nullptr; \
	275	size_t len = 0; \
	276	if (ENCODE(id, REQPTR, &buf, &len) < 0) { _fail("encode failed"); return false; } \
	277	if (!_send_frame(buf, len)) return false; \
	278	const uint8_t *body = nullptr; \
	279	size_t body_len = 0; \
	280	if (!_await(id, timeout_ms, &body, &body_len)) return false; \
	281	if (!body) return false; \
	282	swarmd_ok_t ok; \
	283	memset(&ok, 0, sizeof(ok)); \
	284	compact_state_t in = { 0, body_len, (uint8_t *)body }; \
	285	if (swarmd_ok_decode(&in, &ok) < 0) { _fail("ok reply did not decode"); return false; } \
	286	if (!ok.ok) _err = "the daemon refused the request"; \
	287	return ok.ok; \
	288	} while (0)
	289
	290	bool Client::join(const uint8_t topic[32], bool server, bool client, int timeout_ms) {
	291	swarmd_swarm_join_t req;
	292	memset(&req, 0, sizeof(req));
	293	memcpy(req.topic, topic, 32);
	294	req.server = server;
	295	req.client = client;
	296	SWARMD_OK_CALL(swarmd_encode_swarm_join, &req);
	297	}
	298
	299	bool Client::leave(const uint8_t topic[32], int timeout_ms) {
	300	swarmd_topic_ref_t req;
	301	memset(&req, 0, sizeof(req));
	302	memcpy(req.topic, topic, 32);
	303	SWARMD_OK_CALL(swarmd_encode_swarm_leave, &req);
	304	}
	305
	306	bool Client::close_channel(uint64_t channel, int timeout_ms) {
	307	swarmd_channel_ref_t req;
	308	memset(&req, 0, sizeof(req));
	309	req.channel = channel;
	310	SWARMD_OK_CALL(swarmd_encode_channel_close, &req);
	311	}
	312
	313	bool Client::content_serve(const uint8_t hash[32], const std::string &path, int timeout_ms) {
	314	swarmd_content_serve_t req;
	315	memset(&req, 0, sizeof(req));
	316	memcpy(req.hash, hash, 32);
	317	req.path.data = (utf8_t *)path.c_str();
	318	req.path.len = path.size();
	319	SWARMD_OK_CALL(swarmd_encode_content_serve, &req);
	320	}
	321
	322	bool Client::content_fetch(const uint8_t hash[32], const std::string &dest_path, int timeout_ms) {
	323	swarmd_content_fetch_t req;
	324	memset(&req, 0, sizeof(req));
	325	memcpy(req.hash, hash, 32);
	326	req.dest_path.data = (utf8_t *)dest_path.c_str();
	327	req.dest_path.len = dest_path.size();
	328	SWARMD_OK_CALL(swarmd_encode_content_fetch, &req);
	329	}
	330
	331	bool Client::content_cancel(const uint8_t hash[32], bool discard, int timeout_ms) {
	332	swarmd_content_cancel_t req;
	333	memset(&req, 0, sizeof(req));
	334	memcpy(req.hash, hash, 32);
	335	req.discard = discard;
	336	SWARMD_OK_CALL(swarmd_encode_content_cancel, &req);
	337	}
	338
	339	uint64_t Client::open_channel(const std::string &protocol, const std::vector<uint8_t> &id_bytes,
	340	uint32_t slots, int timeout_ms) {
	341	if (_fd < 0) { _err = "not connected"; return 0; }
	342	swarmd_channel_open_t req;
	343	memset(&req, 0, sizeof(req));
	344	req.protocol.data = (utf8_t *)protocol.c_str();
	345	req.protocol.len = protocol.size();
	346	req.id = id_bytes.empty() ? nullptr : (uint8_t *)id_bytes.data();
	347	req.id_len = id_bytes.size();
	348	req.slots = slots ? slots : 1;
	349
	350	const uint64_t id = _next_id++;
	351	uint8_t *buf = nullptr;
	352	size_t len = 0;
	353	if (swarmd_encode_channel_open(id, &req, &buf, &len) < 0) { _fail("encode failed"); return 0; }
	354	if (!_send_frame(buf, len)) return 0;
	355
	356	const uint8_t *body = nullptr;
	357	size_t body_len = 0;
	358	if (!_await(id, timeout_ms, &body, &body_len) \|\| !body) return 0;
	359
	360	swarmd_channel_ref_t ref;
	361	memset(&ref, 0, sizeof(ref));
	362	compact_state_t in = { 0, body_len, (uint8_t *)body };
	363	if (swarmd_channel_ref_decode(&in, &ref) < 0) { _fail("channel reply did not decode"); return 0; }
	364	if (ref.channel == 0) _err = "the daemon did not open the channel";
	365	return ref.channel;
	366	}
	367
	368	bool Client::send(uint64_t channel, uint32_t slot, uint64_t target,
	369	const uint8_t bytes, size_t len_bytes, bool writable, int timeout_ms) {
	370	if (writable) *writable = false;
	371	if (_fd < 0) { _err = "not connected"; return false; }
	372
	373	swarmd_send_request_t req;
	374	memset(&req, 0, sizeof(req));
	375	req.channel = channel;
	376	req.slot = slot;
	377	req.target = target;
	378	req.bytes = (uint8_t *)bytes;
	379	req.bytes_len = len_bytes;
	380
	381	const uint64_t id = _next_id++;
	382	uint8_t *buf = nullptr;
	383	size_t len = 0;
	384	if (swarmd_encode_send(id, &req, &buf, &len) < 0) { _fail("encode failed"); return false; }
	385	if (!_send_frame(buf, len)) return false;
	386
	387	const uint8_t *body = nullptr;
	388	size_t body_len = 0;
	389	if (!_await(id, timeout_ms, &body, &body_len) \|\| !body) return false;
	390
	391	swarmd_send_result_t res;
	392	memset(&res, 0, sizeof(res));
	393	compact_state_t in = { 0, body_len, (uint8_t *)body };
	394	if (swarmd_send_result_decode(&in, &res) < 0) { _fail("send reply did not decode"); return false; }
	395	if (writable) *writable = res.writable;
	396	if (!res.accepted) _err = "no peer has this channel open yet";
	397	return res.accepted;
	398	}
	399
	400	bool Client::status(Status &out, int timeout_ms) {
	401	if (_fd < 0) { _err = "not connected"; return false; }
	402	swarmd_none_t req;
	403	memset(&req, 0, sizeof(req));
	404	const uint64_t id = _next_id++;
	405	uint8_t *buf = nullptr;
	406	size_t len = 0;
	407	if (swarmd_encode_status_get(id, &req, &buf, &len) < 0) { _fail("encode failed"); return false; }
	408	if (!_send_frame(buf, len)) return false;
	409
	410	const uint8_t *body = nullptr;
	411	size_t body_len = 0;
	412	if (!_await(id, timeout_ms, &body, &body_len) \|\| !body) return false;
	413
	414	swarmd_status_t st;
	415	memset(&st, 0, sizeof(st));
	416	compact_state_t in = { 0, body_len, (uint8_t *)body };
	417	if (swarmd_status_decode(&in, &st) < 0) { _fail("status reply did not decode"); return false; }
	418	out.dht_state = st.dht_state;
	419	out.bootstrapped = st.bootstrapped;
	420	out.peers_known = st.peers_known;
	421	out.peers_connected = st.peers_connected;
	422	out.connecting = st.connecting;
	423	out.queued = st.queued;
	424	out.local_clients = st.local_clients;
	425	return true;
	426	}
	427
	428	bool Client::content_hash(const std::string &path, uint8_t out_hash[32], int timeout_ms) {
	429	memset(out_hash, 0, 32);
	430	if (_fd < 0) { _err = "not connected"; return false; }
	431	swarmd_path_t req;
	432	memset(&req, 0, sizeof(req));
	433	req.path.data = (utf8_t *)path.c_str();
	434	req.path.len = path.size();
	435
	436	const uint64_t id = _next_id++;
	437	uint8_t *buf = nullptr;
	438	size_t len = 0;
	439	if (swarmd_encode_content_hash(id, &req, &buf, &len) < 0) { _fail("encode failed"); return false; }
	440	if (!_send_frame(buf, len)) return false;
	441
	442	const uint8_t *body = nullptr;
	443	size_t body_len = 0;
	444	if (!_await(id, timeout_ms, &body, &body_len) \|\| !body) return false;
	445
	446	swarmd_hash_t res;
	447	memset(&res, 0, sizeof(res));
	448	compact_state_t in = { 0, body_len, (uint8_t *)body };
	449	if (swarmd_hash_decode(&in, &res) < 0) { _fail("hash reply did not decode"); return false; }
	450	memcpy(out_hash, res.hash, 32);
	451	static const uint8_t zero[32] = { 0 };
	452	if (memcmp(out_hash, zero, 32) == 0) {
	453	// The daemon answers all-zero when it cannot read the file. Saying so is
	454	// better than handing back a hash nothing will ever match.
	455	_err = "the daemon could not hash " + path;
	456	return false;
	457	}
	458	return true;
	459	}
	460
	461	} // namespace swarmd

swarmd/client/client.h +153 -0

…	…	@@ -0,0 +1,153 @@
	1	#ifndef SWARMD_CLIENT_H
	2	#define SWARMD_CLIENT_H
	3
	4	// A C++ client for swarmd. docs/SWARMD-DESIGN.md §7 item 7, §9 step 5.
	5	//
	6	// "Any program in C++" is the goal, and until now the only clients were a
	7	// Python test harness and one 250-line tool. Both hand-assembled frames.
	8	//
	9	// ⛔ NOTHING HERE IS HAND-WRITTEN WIRE, and that is the entire reason this file
	10	// exists rather than a page of documentation telling each consumer how to do it
	11	// themselves. Two clients in this repo have already been wrong that way — the
	12	// first channel gate hand-rolled compact-encoding in Python and the daemon
	13	// answered "undecodable message" to every frame, and swarmd-probe gave its sends
	14	// an id range that overlapped the content calls and reported a fetch reply as a
	15	// send result. Both are the same mistake: a client is a wire implementation, and
	16	// there should be ONE.
	17	//
	18	// SHAPE. Setup calls block for their reply; everything the daemon says on its
	19	// own arrives through callbacks during poll(). That split is not arbitrary:
	20	// channel_open has to return a handle before you can send on it, while a peer
	21	// arriving has nobody waiting for it.
	22	//
	23	// ⛔ AND A BLOCKING CALL STILL DISPATCHES EVENTS. Frames do not queue politely
	24	// behind the reply you asked for: a peer can come up, or a 100 MB fetch can
	25	// finish, while you are waiting for the answer to channel_open. Anything read on
	26	// the way to a reply is delivered, not dropped and not deferred.
	27	//
	28	// THREADING. One Client belongs to one thread. It owns a socket and a buffer and
	29	// guards neither.
	30
	31	#include <cstddef>
	32	#include <cstdint>
	33	#include <functional>
	34	#include <map>
	35	#include <string>
	36	#include <vector>
	37
	38	namespace swarmd {
	39
	40	// What the daemon says on its own. Every one of these arrives inside poll(), or
	41	// inside a blocking call that read it on the way to a reply.
	42	struct ClientCallbacks {
	43	// A frame from a peer, on a channel this client opened.
	44	std::function<void(uint64_t channel, uint64_t slot, uint64_t origin_kind,
	45	uint64_t origin_id, const uint8_t *bytes, size_t len)> on_message;
	46	std::function<void(uint64_t peer, const uint8_t pubkey[32])> on_peer_up;
	47	std::function<void(uint64_t peer)> on_peer_down;
	48	// The channel has drained and will accept writes again.
	49	std::function<void(uint64_t channel)> on_drain;
	50
	51	std::function<void(const uint8_t hash[32], uint64_t received, uint64_t total)> on_content_progress;
	52	std::function<void(const uint8_t hash[32], const std::string &path)> on_content_done;
	53	std::function<void(const uint8_t hash[32], const std::string &reason)> on_content_failed;
	54
	55	// The daemon went away, or sent something this client could not read. After
	56	// this the Client is closed; connect() again to resume.
	57	std::function<void(const std::string &reason)> on_disconnect;
	58	};
	59
	60	struct Status {
	61	uint64_t dht_state = 0; // 0 starting, 1 connecting, 2 up
	62	bool bootstrapped = false;
	63	uint64_t peers_known = 0;
	64	uint64_t peers_connected = 0;
	65	uint64_t connecting = 0;
	66	uint64_t queued = 0;
	67	uint64_t local_clients = 0;
	68	};
	69
	70	class Client {
	71	public:
	72	Client() = default;
	73	~Client();
	74	Client(const Client &) = delete;
	75	Client &operator=(const Client &) = delete;
	76
	77	ClientCallbacks on;
	78
	79	// Dial the daemon and say HELLO. `name` and `app_identity` are what the
	80	// daemon records for this client; app_identity is what echo suppression and
	81	// membership will key on, so a program that has one should pass it.
	82	bool connect(const std::string &socket_path, const std::string &name,
	83	const uint8_t app_identity[32] = nullptr, int timeout_ms = 5000);
	84	void close();
	85	bool connected() const { return _fd >= 0; }
	86
	87	uint64_t client_id() const { return _client_id; }
	88	const uint8_t *daemon_pubkey() const { return _daemon_pubkey; }
	89	const std::string &last_error() const { return _err; }
	90
	91	// Serve/receive on a topic. Both flags default true, which is what a peer
	92	// that wants to meet anyone on the topic wants.
	93	bool join(const uint8_t topic[32], bool server = true, bool client = true,
	94	int timeout_ms = 5000);
	95	bool leave(const uint8_t topic[32], int timeout_ms = 5000);
	96
	97	// Returns a channel handle, or 0. `slots` is how many message types the
	98	// channel carries.
	99	uint64_t open_channel(const std::string &protocol,
	100	const std::vector<uint8_t> &id = {},
	101	uint32_t slots = 1, int timeout_ms = 5000);
	102	bool close_channel(uint64_t channel, int timeout_ms = 5000);
	103
	104	// ⛔ `accepted` FALSE MEANS THE FRAME WAS NOT TAKEN — retry, do not assume.
	105	// The daemon refuses when the handle is unknown or the channel is open on no
	106	// peer, which is every moment before the remote opens its side. `writable`
	107	// says whether more will be accepted right now; on false, wait for on_drain.
	108	// `target` 0 means every peer holding this channel.
	109	bool send(uint64_t channel, uint32_t slot, uint64_t target,
	110	const uint8_t bytes, size_t len, bool writable = nullptr,
	111	int timeout_ms = 5000);
	112
	113	bool status(Status &out, int timeout_ms = 5000);
	114
	115	// ---- content ----
	116	// The overlay's hash of a file the DAEMON can read — blake2b over the
	117	// hypercore leaf construction. A hash from anywhere else is one the overlay
	118	// will reject, which is why this asks the daemon rather than computing it.
	119	bool content_hash(const std::string &path, uint8_t out_hash[32], int timeout_ms = 30000);
	120	bool content_serve(const uint8_t hash[32], const std::string &path, int timeout_ms = 5000);
	121	// Returns once the fetch has STARTED. Watch on_content_done / _failed.
	122	bool content_fetch(const uint8_t hash[32], const std::string &dest_path, int timeout_ms = 5000);
	123	bool content_cancel(const uint8_t hash[32], bool discard, int timeout_ms = 5000);
	124
	125	// Read whatever is there and dispatch it. Returns false once disconnected.
	126	// `timeout_ms` 0 polls without waiting.
	127	bool poll(int timeout_ms);
	128
	129	private:
	130	// Send one encoded request. Takes ownership of `buf` exactly as the
	131	// generated encoders hand it over: heap-allocated, caller frees.
	132	bool _send_frame(uint8_t *buf, size_t len);
	133	// Pump until the response to `id` lands, dispatching events on the way.
	134	// `body`/`body_len` point into `_reply` and are valid until the next pump.
	135	bool _await(uint64_t id, int timeout_ms, const uint8_t *body, size_t body_len);
	136	bool _read_some(int timeout_ms);
	137	// Consume whole frames from `_in`. Returns true if the awaited id landed.
	138	bool _drain_frames(uint64_t await_id, const uint8_t *body, size_t body_len);
	139	void _dispatch_event(const void *msg);
	140	void _fail(const std::string &why);
	141
	142	int _fd = -1;
	143	uint64_t _next_id = 1;
	144	uint64_t _client_id = 0;
	145	uint8_t _daemon_pubkey[32] = { 0 };
	146	std::string _err;
	147	std::vector<uint8_t> _in; // partial inbound
	148	std::vector<uint8_t> _reply; // the frame a blocking call is returning
	149	};
	150
	151	} // namespace swarmd
	152
	153	#endif

swarmd/test/test_client.cpp +204 -0

…	…	@@ -0,0 +1,204 @@
	1	// Gate: the C++ client library, against a real Daemon over a real socket.
	2	//
	3	// docs/SWARMD-DESIGN.md §9 step 5. Until now the wire had two client
	4	// implementations — a Python harness and swarmd-probe — and both were wrong at
	5	// least once in ways the daemon could not distinguish from a bad daemon. This
	6	// asserts the one that replaces them.
	7	//
	8	// ⛔ NO NETWORK. The daemon is pointed at a bootstrap that answers nothing, so
	9	// it never has a peer. Everything here is about the LOCAL contract: the calls,
	10	// their replies, and the events. Peer-to-peer behaviour is
	11	// test_channel_e2e.py's and test_content_e2e.py's job, on the real DHT.
	12	//
	13	// The daemon runs on its own thread because that is how a daemon runs. A client
	14	// that blocks for a reply has nobody to pump the thing it is waiting for
	15	// otherwise, and pretending otherwise is what a single-threaded gate would be
	16	// asserting instead.
	17
	18	#include <atomic>
	19	#include <cstdio>
	20	#include <cstdlib>
	21	#include <cstring>
	22	#include <string>
	23	#include <chrono>
	24	#include <thread>
	25	#include <vector>
	26
	27	#include <unistd.h>
	28	#include <sys/stat.h>
	29	#include <cerrno>
	30
	31	#include "client.h"
	32	#include "daemon.h"
	33
	34	static int checks = 0, failures = 0;
	35	static void check(bool c, const char *what) {
	36	checks++;
	37	if (c) { printf(" PASS: %s\n", what); return; }
	38	failures++;
	39	printf(" FAIL: %s\n", what);
	40	}
	41
	42	int main() {
	43	char sock[128], file[128];
	44	snprintf(sock, sizeof(sock), "/tmp/swarmd-client-%d.sock", (int)getpid());
	45	snprintf(file, sizeof(file), "/tmp/swarmd-client-%d.bin", (int)getpid());
	46
	47	// A file for the content calls to name. Small: this gate is about the
	48	// contract, not about throughput.
	49	{
	50	FILE *f = fopen(file, "wb");
	51	if (!f) { printf("cannot write %s\n", file); return 1; }
	52	std::vector<uint8_t> bytes(64 * 1024);
	53	for (size_t i = 0; i < bytes.size(); i++) bytes[i] = (uint8_t)(i * 31 + 7);
	54	fwrite(bytes.data(), 1, bytes.size(), f);
	55	fclose(f);
	56	}
	57
	58	char store[128];
	59	snprintf(store, sizeof(store), "/tmp/swarmd-client-store-%d", (int)getpid());
	60	if (mkdir(store, 0755) != 0 && errno != EEXIST) { printf("cannot make %s\n", store); return 1; }
	61
	62	swarmd::Daemon d;
	63	if (!d.start(sock, "127.0.0.1:1", store)) { printf("daemon start failed\n"); return 1; }
	64
	65	std::atomic<bool> stop(false);
	66	std::thread pump([&d, &stop]() { while (!stop.load()) d.poll(5); });
	67
	68	int rc = 0;
	69	{
	70	swarmd::Client c;
	71
	72	// Events the daemon raises on its own.
	73	int content_failed = 0;
	74	std::string last_fail;
	75	int failed_during_status = 0;
	76	bool counting = false;
	77	c.on.on_content_failed = [&](const uint8_t *, const std::string &why) {
	78	content_failed++;
	79	last_fail = why;
	80	if (counting) failed_during_status++;
	81	};
	82	int disconnects = 0;
	83	c.on.on_disconnect = [&](const std::string &) { disconnects++; };
	84
	85	uint8_t identity[32];
	86	memset(identity, 0x5c, 32);
	87
	88	check(c.connect(sock, "gate", identity), "connect() reached the daemon and said HELLO");
	89	check(c.connected(), "and reports itself connected");
	90
	91	// ⛔ THESE TWO ARE THE POINT OF HELLO. §5 says the reply carries
	92	// {client_id, daemon_pubkey}; §7 item 4 says client ids exist so a client
	93	// can suppress its own echo. A client that cannot learn its own id cannot
	94	// do that, and a zeroed daemon key tells it nothing about who it reached.
	95	check(c.client_id() != 0, "HELLO answered with this client's OWN id");
	96	{
	97	static const uint8_t zero[32] = { 0 };
	98	check(memcmp(c.daemon_pubkey(), zero, 32) != 0,
	99	"and with the daemon's real public key, not 32 zeroes");
	100	}
	101
	102	uint8_t topic[32];
	103	memset(topic, 0x3a, 32);
	104	check(c.join(topic), "join() was accepted");
	105
	106	const uint64_t ch = c.open_channel("client-gate/v1", {}, 2);
	107	check(ch != 0, "open_channel() returned a handle");
	108
	109	// The send-truth contract, now through the library: with no peer there
	110	// is nowhere to put the frame, and saying so is the whole point.
	111	bool writable = true;
	112	const char *payload = "nowhere-to-go";
	113	check(!c.send(ch, 0, 0, (const uint8_t *)payload, strlen(payload), &writable),
	114	"send() on a channel no peer has open is REFUSED, not silently dropped");
	115	check(!writable, "and the channel is not reported writable");
	116
	117	swarmd::Status st;
	118	check(c.status(st), "status() answered");
	119	check(st.local_clients >= 1, "and counts this client");
	120	check(st.peers_connected == 0, "and reports no peers, which is the truth here");
	121
	122	// ---- content ----
	123	uint8_t hash[32], hash2[32];
	124	check(c.content_hash(file, hash), "content_hash() hashed a real file");
	125	check(c.content_hash(file, hash2) && memcmp(hash, hash2, 32) == 0,
	126	"and the same bytes hash the same way twice");
	127	{
	128	static const uint8_t zero[32] = { 0 };
	129	check(memcmp(hash, zero, 32) != 0, "and the hash is not all zeroes");
	130	}
	131	check(!c.content_hash("/nonexistent/path/at/all", hash2),
	132	"hashing a file that is not there FAILS rather than returning zeroes");
	133
	134	check(c.content_serve(hash, file), "content_serve() registered the file");
	135
	136	// Cancelling a fetch nobody started is refused by liboverlay, and the
	137	// daemon reports that refusal as a FAILED event rather than swallowing it.
	138	uint8_t missing[32];
	139	memset(missing, 0x77, 32);
	140	check(c.content_cancel(missing, false),
	141	"content_cancel() was accepted by the daemon (the RPC, not the outcome)");
	142
	143	// ⛔ THE INTERLEAVE — and the wait is what MAKES it one, not a patch over
	144	// a flake.
	145	//
	146	// The cancel's ok reply is written by the control handler synchronously;
	147	// the FAILED event is produced later, on the UDX loop, and broadcast by a
	148	// subsequent Daemon::poll. Asserting straight after the cancel asserts a
	149	// race: status() is answered just as fast, and the first version of this
	150	// gate failed against a client that was working. Waiting until the event
	151	// is known to be sitting in our socket is what establishes the condition
	152	// the assertion is about — a frame queued AHEAD of the reply a blocking
	153	// call is waiting for. A client that drops what it reads on the way to
	154	// its own answer loses it, invisibly: no error, just an event that never
	155	// fires.
	156	{
	157	const uint64_t deadline = 3000;
	158	uint64_t waited = 0;
	159	while (content_failed == 0 && waited < deadline) {
	160	std::this_thread::sleep_for(std::chrono::milliseconds(20));
	161	waited += 20;
	162	// Deliberately NOT polling the client: the point is to leave the
	163	// frame unread so the next blocking call has to handle it.
	164	}
	165	// It is in our socket now, unread, because nothing has read it.
	166	std::this_thread::sleep_for(std::chrono::milliseconds(50));
	167	}
	168	counting = true;
	169	check(c.status(st), "a second status() answered");
	170	counting = false;
	171	check(content_failed >= 1, "the daemon reported the refused cancel at all");
	172	check(failed_during_status >= 1,
	173	"an event queued AHEAD of a reply was delivered by the blocking call "
	174	"that read past it, not dropped");
	175	check(last_fail.find("no such fetch") != std::string::npos,
	176	"and it is the event the daemon actually sent");
	177
	178	check(c.close_channel(ch), "close_channel() was accepted");
	179	check(disconnects == 0, "nothing in that sequence looked like a disconnect");
	180
	181	// ---- what a closed client does ----
	182	c.close();
	183	check(!c.connected(), "close() closes");
	184	check(!c.join(topic), "and a call on a closed client fails rather than crashing");
	185	check(c.last_error() == "not connected", "with an error that says why");
	186
	187	// ---- a daemon that is not there ----
	188	swarmd::Client absent;
	189	check(!absent.connect("/tmp/swarmd-does-not-exist.sock", "gate"),
	190	"connecting to a socket with no daemon fails");
	191	check(absent.last_error().find("no daemon at") != std::string::npos,
	192	"and says so");
	193
	194	rc = failures ? 1 : 0;
	195	}
	196
	197	stop.store(true);
	198	pump.join();
	199	d.stop();
	200	unlink(file);
	201
	202	printf("\n%d/%d passed\n", checks - failures, checks);
	203	return rc;
	204	}

hyperswarm-peer commits / 54eb4c5

Changed files (5) +829 -1