#ifndef SWARMD_UDS_SERVER_H
#define SWARMD_UDS_SERVER_H
// The local socket, and the per-client buffering session-daemon does not have.
//
// ⛔ THE BUG THIS EXISTS TO NOT REPEAT. session_daemon.cpp fans out to clients
// 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 client's
// stream. The client then reads a garbage length, hits its own frame cap and
// closes. A slow reader silently corrupts its own session, and the daemon calls
// that success.
//
// So: every client owns an output buffer, a partial write is remembered rather
// than discarded, and a client that exceeds its high-water mark is DROPPED —
// because a dropped client reconnects and resyncs, while a corrupted one does
// not know it is corrupted.
//
// Framing is uint32 LE length + payload, matching session-daemon and
// blender-bridge/session/client.py so an existing client keeps working.
#include <cstddef>
#include <cstdint>
#include <functional>
#include <map>
#include <string>
#include <vector>
namespace swarmd {
// A client that has this much unsent output is not keeping up.
static constexpr size_t UDS_CLIENT_HIGH_WATER = 8u * 1024 * 1024;
// A length this large is a framing bug, not a frame.
static constexpr uint32_t UDS_MAX_FRAME = 8u * 1024 * 1024;
struct UdsClient {
int fd = -1;
uint64_t id = 0;
std::vector<uint8_t> in; // partial inbound frame
std::vector<uint8_t> out; // unsent outbound bytes
size_t out_sent = 0; // how much of out[0] has gone
bool writable = true; // false between high-water and drain
};
class UdsServer {
public:
~UdsServer();
bool listen(const char *socket_path);
void close_all();
// Drive accept/read/write once. `timeout_ms` bounds the poll.
void poll(int timeout_ms);
// One whole frame arrived from a client.
std::function<void(uint64_t client_id, const uint8_t *data, size_t len)> on_frame;
// A client attached or went away.
std::function<void(uint64_t client_id)> on_connect;
std::function<void(uint64_t client_id)> on_disconnect;
// Queue a frame. Returns false if the client is unknown or was dropped for
// exceeding its high-water mark. `r_writable` reports whether the client is
// still under that mark — the caller is expected to stop producing when it
// is not, which is the answer Protomux gives and session-daemon discards.
bool send(uint64_t client_id, const uint8_t *data, size_t len, bool *r_writable = nullptr);
// Queue to every client except `except_id` (0 = all).
void broadcast(const uint8_t *data, size_t len, uint64_t except_id = 0);
int client_count() const { return (int)_clients.size(); }
uint64_t frames_in() const { return _frames_in; }
uint64_t frames_out() const { return _frames_out; }
uint64_t dropped_clients() const { return _dropped; }
private:
void _accept();
void _read(UdsClient &c);
void _flush(UdsClient &c);
void _drop(uint64_t id);
int _listen_fd = -1;
std::string _socket_path;
std::map<uint64_t, UdsClient> _clients;
std::vector<uint64_t> _dead; // dropped during a poll, reaped after it
uint64_t _next_id = 1;
uint64_t _frames_in = 0;
uint64_t _frames_out = 0;
uint64_t _dropped = 0;
};
} // namespace swarmd
#endif