// Gate D2 + D3 (docs/SWARMD-DESIGN.md §8): a client that stops reading is
// DROPPED, and the survivors' streams stay intact.
//
// ⛔ THIS IS THE ASSERTION session-daemon FAILS. It fans out from the UDX thread
// with unchecked writes on non-blocking fds:
//
// if (write(fd, hdr, 4) != 4) continue;
// if (write(fd, data, len) != (ssize_t)len) continue;
//
// A partial write there does not drop a frame, it desynchronises the stream —
// the client reads a garbage length next and closes. A slow reader silently
// corrupts its own session while the daemon reports success.
#include <cstdio>
#include <cstring>
#include <string>
#include <vector>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#include <fcntl.h>
#include "uds_server.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;
}
int main() {
char path[128];
snprintf(path, sizeof(path), "/tmp/swarmd-gate-%d.sock", (int)getpid());
swarmd::UdsServer s;
if (!s.listen(path)) { printf("listen failed\n"); return 1; }
int reader = dial(path);
int sulker = dial(path); // connects, then never reads a byte
check(reader >= 0 && sulker >= 0, "two clients attached");
s.poll(50);
check(s.client_count() == 2, "the server sees both");
uint64_t reader_id = 0, sulker_id = 0;
{
// Identify them by which fd answers: send a probe to each id in turn.
// Ids are handed out in accept order, and `reader` dialled first.
reader_id = 1;
sulker_id = 2;
}
// Fill the sulker past its high-water mark. It never reads, so the kernel
// buffer fills, writes go EAGAIN, and the output buffer grows.
std::vector<uint8_t> chunk(64 * 1024, 0xab);
bool sulker_dropped = false;
bool last_writable = true;
for (int i = 0; i < 4096 && !sulker_dropped; i++) {
bool w = true;
if (!s.send(sulker_id, chunk.data(), chunk.size(), &w)) sulker_dropped = true;
else last_writable = w;
s.poll(0);
}
check(sulker_dropped, "a client that never reads is eventually DROPPED");
check(s.dropped_clients() >= 1, "and the drop is counted");
check(!last_writable || sulker_dropped,
"writability went false before the drop (D3)");
// ⛔ THE ONE THAT MATTERS. The survivor's stream must still be frame-aligned.
const char *msg = "intact";
check(s.send(reader_id, (const uint8_t *)msg, strlen(msg)),
"the surviving client still accepts a frame");
for (int i = 0; i < 20; i++) s.poll(5);
uint8_t buf[256];
ssize_t got = ::read(reader, buf, sizeof(buf));
bool aligned = false;
if (got >= 4) {
const uint32_t n = (uint32_t)buf[0] | ((uint32_t)buf[1] << 8) |
((uint32_t)buf[2] << 16) | ((uint32_t)buf[3] << 24);
aligned = (n == strlen(msg)) &&
(got >= (ssize_t)(4 + n)) &&
(memcmp(buf + 4, msg, n) == 0);
}
check(aligned, "the survivor's stream is still frame-aligned, not corrupted");
close(reader);
if (!sulker_dropped) close(sulker);
s.close_all();
printf("\n%d/%d passed\n", checks - failures, checks);
return failures ? 1 : 0;
}