// Gate: SEND must not claim a frame was accepted when there is nowhere to put it.
//
// The daemon's send backend crosses to the UDX thread with run_on_loop, so it
// cannot know synchronously that protomux took the bytes. What it CAN know is
// whether the channel is open on any peer at all — and if it is open on none,
// `accepted` and `writable` are answers about a frame that will be dropped by
// ProtomuxChannel::send (which returns false when !opened) with nobody told.
//
// ⛔ THIS IS THE SHAPE OF #72's WALL. A path that fails with a bare return and
// reports success is not a smaller bug because the failure is rarer; it is a
// bigger one, because the caller has been told to stop looking.
//
// No network: the daemon is pointed at a bootstrap that does not exist, so it
// never has a peer and the answer is unambiguous.
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <vector>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#include "daemon.h"
extern "C" {
#include <compact.h>
#include <rpc.h>
}
static int checks = 0, failures = 0;
static void check(bool c, const char *what) {
checks++;
if (c) { printf(" PASS: %s\n", what); return; }
failures++;
printf(" FAIL: %s\n", what);
}
static int dial(const char *path) {
int fd = socket(AF_UNIX, SOCK_STREAM, 0);
struct sockaddr_un a;
memset(&a, 0, sizeof(a));
a.sun_family = AF_UNIX;
strncpy(a.sun_path, path, sizeof(a.sun_path) - 1);
if (connect(fd, (struct sockaddr *)&a, sizeof(a)) != 0) { close(fd); return -1; }
return fd;
}
static void write_frame(int fd, uint8_t *d, size_t n) {
uint8_t hdr[4] = { (uint8_t)(n & 0xff), (uint8_t)((n >> 8) & 0xff),
(uint8_t)((n >> 16) & 0xff), (uint8_t)((n >> 24) & 0xff) };
(void)!write(fd, hdr, 4);
(void)!write(fd, d, n);
free(d); // the caller frees what *_encode_* allocated
}
// Pump the daemon until a frame carrying a response to `want_id` arrives.
static bool await_reply(swarmd::Daemon &d, int fd, uint64_t want_id, rpc_message_t *out,
std::vector<uint8_t> &store) {
for (int i = 0; i < 400; i++) {
d.poll(5);
uint8_t hdr[4];
ssize_t r = recv(fd, hdr, 4, MSG_DONTWAIT | MSG_PEEK);
if (r != 4) continue;
const uint32_t n = (uint32_t)hdr[0] | ((uint32_t)hdr[1] << 8) |
((uint32_t)hdr[2] << 16) | ((uint32_t)hdr[3] << 24);
std::vector<uint8_t> buf(4 + n);
size_t got = 0;
while (got < buf.size()) {
r = read(fd, buf.data() + got, buf.size() - got);
if (r <= 0) break;
got += (size_t)r;
}
if (got != buf.size()) continue;
store.assign(buf.begin() + 4, buf.end());
compact_state_t st = { 0, store.size(), store.data() };
if (rpc_decode_message(&st, out) < 0) continue;
if (out->type == rpc_response && out->id == want_id) return true;
}
return false;
}
int main() {
char path[128];
snprintf(path, sizeof(path), "/tmp/swarmd-sendtruth-%d.sock", (int)getpid());
swarmd::Daemon d;
// A bootstrap that answers nothing. The daemon runs; it never gets a peer.
if (!d.start(path, "127.0.0.1:1", nullptr)) { printf("start failed\n"); return 1; }
int fd = dial(path);
check(fd >= 0, "client attached");
d.poll(20);
uint8_t *buf = nullptr;
size_t len = 0;
swarmd_hello_request_t hello;
memset(&hello, 0, sizeof(hello));
hello.name.data = (utf8_t *)"gate";
hello.name.len = 4;
hello.version = 1;
if (swarmd_encode_hello(1, &hello, &buf, &len) >= 0) write_frame(fd, buf, len);
rpc_message_t msg;
std::vector<uint8_t> store;
check(await_reply(d, fd, 1, &msg, store), "HELLO was answered");
// ---- open a channel while no peer exists ----
swarmd_channel_open_t co;
memset(&co, 0, sizeof(co));
co.protocol.data = (utf8_t *)"sendtruth/v1";
co.protocol.len = 12;
co.id = nullptr; co.id_len = 0;
co.slots = 1;
if (swarmd_encode_channel_open(2, &co, &buf, &len) >= 0) write_frame(fd, buf, len);
uint64_t handle = 0;
if (await_reply(d, fd, 2, &msg, store)) {
swarmd_channel_ref_t ref;
memset(&ref, 0, sizeof(ref));
compact_state_t in = { 0, msg.len, msg.data };
if (swarmd_channel_ref_decode(&in, &ref) >= 0) handle = ref.channel;
}
check(handle != 0, "the channel was opened and has a handle");
// Let the run_on_loop that creates the spec actually run.
for (int i = 0; i < 60; i++) d.poll(5);
// ---- send into it ----
const char *payload = "into-the-void";
swarmd_send_request_t sr;
memset(&sr, 0, sizeof(sr));
sr.channel = handle;
sr.slot = 0;
sr.target = 0;
sr.bytes = (uint8_t *)payload;
sr.bytes_len = strlen(payload);
if (swarmd_encode_send(3, &sr, &buf, &len) >= 0) write_frame(fd, buf, len);
swarmd_send_result_t res;
memset(&res, 0, sizeof(res));
bool got_reply = false;
if (await_reply(d, fd, 3, &msg, store)) {
compact_state_t in = { 0, msg.len, msg.data };
got_reply = swarmd_send_result_decode(&in, &res) >= 0;
}
check(got_reply, "SEND was answered");
check(!res.accepted,
"a send with no peer channel open is NOT reported as accepted");
check(!res.writable,
"and the channel is NOT reported writable");
// ---- and an unknown handle is refused just as plainly ----
memset(&sr, 0, sizeof(sr));
sr.channel = handle + 999;
sr.bytes = (uint8_t *)payload;
sr.bytes_len = strlen(payload);
if (swarmd_encode_send(4, &sr, &buf, &len) >= 0) write_frame(fd, buf, len);
memset(&res, 0, sizeof(res));
res.accepted = true;
if (await_reply(d, fd, 4, &msg, store)) {
compact_state_t in = { 0, msg.len, msg.data };
swarmd_send_result_decode(&in, &res);
}
check(!res.accepted, "a send on a handle that does not exist is not accepted");
close(fd);
d.stop();
printf("\n%d/%d passed\n", checks - failures, checks);
return failures ? 1 : 0;
}