Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 16 additions & 4 deletions docs/api.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,24 @@ The full public API is declared in [`include/bacaro.h`](../include/bacaro.h).
## Lifecycle

```c
bacaro_t *bacaro_new (const char *name);
bacaro_t *bacaro_new (const char *name, const char **published_domains);
void bacaro_destroy(bacaro_t **self);
```

`bacaro_new` creates a new instance with the given process name, binds its sockets, and starts peer discovery. Returns `NULL` on failure.

`bacaro_destroy` disconnects all peers, removes IPC files, and frees all resources. Sets `*self` to `NULL`.
`published_domains` is an optional NULL-terminated array of domain strings that this process will publish. When provided, a `.manifest` file is written alongside the `.pub` and `.rep` files so that peers can skip snapshot requests for non-overlapping domains. Pass `NULL` if you don't want to declare a manifest — the bus works identically without one.

```c
// Declare published domains (optional optimisation)
const char *domains[] = {"sensors", "power", NULL};
bacaro_t *b = bacaro_new("powerd", domains);

// No manifest — backwards compatible
bacaro_t *b = bacaro_new("monitor", NULL);
```

`bacaro_destroy` disconnects all peers, removes IPC files (including the `.manifest` if one was written), and frees all resources. Sets `*self` to `NULL`.

## Subscriptions

Expand Down Expand Up @@ -108,7 +119,8 @@ typedef void (*bacaro_fn)(bacaro_t *self, const char *path,
#include <msgpack.hpp>

// Publisher
bacaro_t *pub = bacaro_new("powerd");
const char *domains[] = {"system", "power", nullptr};
bacaro_t *pub = bacaro_new("powerd", domains);

msgpack::sbuffer buf;
msgpack::pack(buf, 87.3);
Expand All @@ -124,7 +136,7 @@ static void on_update(bacaro_t *self, const char *path,
// called on every received update
}

bacaro_t *sub = bacaro_new("monitor");
bacaro_t *sub = bacaro_new("monitor", nullptr);
bacaro_subscribe(sub, "system");
bacaro_on_update(sub, on_update, NULL);

Expand Down
23 changes: 13 additions & 10 deletions docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -29,17 +29,18 @@ Each process uses a **single shared SUB socket** that connects to every peer's P
1. Generate a UUID for this instance.
2. Resolve the runtime directory (`BACARO_RUNTIME_DIR`, default `/tmp/bacaro`).
3. Create the runtime directory if it does not exist.
4. Bind a `ZMQ_PUB` socket → IPC file `<name>.<uuid>.pub`
5. Bind a `ZMQ_ROUTER` socket → IPC file `<name>.<uuid>.rep`
6. Call `discovery_init` (see Discovery below).
4. If `published_domains` is non-NULL, write a `.manifest` file (`<name>.<uuid>.manifest`) listing one domain per line — written **before** discovery so peers see it when they find the `.pub` file.
5. Bind a `ZMQ_PUB` socket → IPC file `<name>.<uuid>.pub`
6. Bind a `ZMQ_ROUTER` socket → IPC file `<name>.<uuid>.rep`
7. Call `discovery_init` (see Discovery below).

The UUID in the filename prevents collisions when a process restarts quickly under the same name.

### Shutdown (`bacaro_destroy`)

1. `discovery_cleanup`: disconnect all peers, close epoll and inotify fds.
2. Set `ZMQ_LINGER = 200ms` on the PUB socket before closing, so any recently published messages have time to flush to connected peers (see [Slow-Joiner Note](#slow-joiner-note)).
3. Close and remove the `.pub` and `.rep` IPC files.
3. Close and remove the `.pub`, `.rep`, and `.manifest` IPC files.
4. Destroy the ZMQ context.

---
Expand Down Expand Up @@ -69,10 +70,9 @@ Triggered by a new `.pub` file appearing (either from the initial scan or from a
1. Extract the peer name from the filename (everything before the first `.`).
2. Derive the `.rep` filename by replacing the `.pub` suffix.
3. Call `zmq_connect` on the **shared SUB socket** to the peer's PUB endpoint.
4. Create a per-peer `ZMQ_DEALER` socket, connect to `ipc://<runtime_dir>/<rep_filename>`.
5. Register the DEALER socket's ZMQ fd with epoll.
6. Record the peer in `peers` and `dealer_fd_to_filename` maps. Store the PUB endpoint string for later `zmq_disconnect`.
7. Send a snapshot request for each currently subscribed domain prefix.
4. Read the peer's `.manifest` file if present; store declared domains and `has_manifest` flag in `PeerInfo`.
5. Record the peer in the `peers` map, storing both the PUB and REP endpoints.
6. **Lazy DEALER**: if the process has active subscriptions, iterate them. For each prefix that overlaps with the peer's manifest (or if the peer has no manifest), create a per-peer `ZMQ_DEALER` socket, connect to the REP endpoint, register its fd with epoll, and send a snapshot request. Peers with a manifest that doesn't overlap any active subscription get no DEALER socket at all.

### Peer disconnect (`discovery_peer_disconnect`)

Expand Down Expand Up @@ -257,7 +257,10 @@ The main handle. One per process. Contains:
### `PeerInfo`

Per-peer connection state:
- `dealer_sock` — per-peer ZMQ DEALER socket for snapshot protocol
- `dealer_sock` — per-peer ZMQ DEALER socket for snapshot protocol (created lazily, may be `nullptr`)
- `name` — peer process name (extracted from filename)
- `pub_endpoint` — stored for `zmq_disconnect` on the shared SUB socket
- `dealer_fd` — ZMQ fd registered with epoll
- `rep_endpoint` — stored for lazy DEALER connect
- `dealer_fd` — ZMQ fd registered with epoll (`-1` until DEALER is created)
- `manifest` — declared published domains read from the peer's `.manifest` file
- `has_manifest` — `true` if a `.manifest` file was found at discovery time
31 changes: 24 additions & 7 deletions docs/concepts.md
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,9 @@ There is no central broker. Each process:

1. Binds a `ZMQ_PUB` socket and a `ZMQ_ROUTER` socket, creating IPC files in the runtime directory.
2. Watches the runtime directory with **inotify** for new peers.
3. Connects a `ZMQ_SUB` socket and a `ZMQ_DEALER` socket to each peer on discovery.
4. Requests a **snapshot** of current state from each peer on connect, seeding its local cache.
3. Connects a single shared `ZMQ_SUB` socket to each peer's PUB endpoint on discovery.
4. Creates a per-peer `ZMQ_DEALER` socket lazily — only when a subscription overlaps with the peer's manifest (or when the peer has no manifest).
5. Requests a **snapshot** of current state from each relevant peer on connect, seeding its local cache.

After the initial snapshot, live `PUB/SUB` updates keep every subscriber's cache current.

Expand All @@ -33,7 +34,7 @@ Any process can publish on any path at any time. There is no ownership enforceme

## Publisher identity

No identity frame is added to the wire format. The publisher's name is inferred from which IPC socket the message arrived on — zero overhead.
The publisher's name is carried explicitly as a frame in the wire format. This is necessary because all messages from all peers arrive on a single shared SUB socket — there is no per-peer socket to infer identity from.

## Late join / snapshot protocol

Expand All @@ -43,14 +44,30 @@ Bacaro solves this automatically: every time a new peer is discovered or a new s

This means `bacaro_get` always returns the latest known value immediately, without any blocking network call.

## Manifest

An optional optimisation for deployments where the set of processes and their published domains is known at build time.

When calling `bacaro_new`, pass a NULL-terminated array of domain strings that the process will publish:

```c
const char *domains[] = {"sensors", "power", NULL};
bacaro_t *b = bacaro_new("powerd", domains);
```

Bacaro writes this list to a `.manifest` file alongside the `.pub` file. When peers discover the process, they read the manifest and only open a DEALER socket (and request a snapshot) if their subscriptions overlap with the declared domains. Peers with no overlapping subscriptions skip the handshake entirely, reducing boot-time connection overhead.

Passing `NULL` disables the manifest — the bus works identically without one. The manifest is an optimisation hint, not access control: a process can still publish on any path regardless of what it declared.

## Wire format

Every published message is a three-frame ZMQ multipart message:
Every published message is a four-frame ZMQ multipart message:

```
Frame 0 — topic : "domain.sub.property" (ZMQ SUB prefix filter)
Frame 1 — header : version(1) | flags(1) | sequence(8) | timestamp(8)
Frame 2 — payload : MessagePack bytes
Frame 0 — topic : "domain.sub.property" (ZMQ SUB prefix filter)
Frame 1 — publisher : publisher process name
Frame 2 — header : version(1) | flags(1) | sequence(8) | timestamp(8)
Frame 3 — payload : MessagePack bytes
```

Snapshot request/reply uses the same framing over the `DEALER/ROUTER` pair, with a flag byte indicating request, reply, or end-of-snapshot.
Expand Down
8 changes: 6 additions & 2 deletions include/bacaro.h
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,12 @@ typedef void (*bacaro_fn)(bacaro_t *self,
const uint8_t *data, size_t len,
void *userdata);

/* Lifecycle */
bacaro_t *bacaro_new (const char *name);
/* Lifecycle.
published_domains is an optional NULL-terminated array of domain prefixes
this process will publish. If provided, a manifest file is written so peers
can skip snapshot requests for non-overlapping domains. Pass NULL to opt out
(all peers will snapshot as usual). */
bacaro_t *bacaro_new (const char *name, const char **published_domains);
void bacaro_destroy(bacaro_t **self);

/* Subscriptions */
Expand Down
2 changes: 1 addition & 1 deletion llms.txt
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ Optional build flags (CMake option or env variable):

```c
// Lifecycle
bacaro_t *bacaro_new(const char *name);
bacaro_t *bacaro_new(const char *name, const char **published_domains);
void bacaro_destroy(bacaro_t **self);

// Subscriptions
Expand Down
25 changes: 23 additions & 2 deletions src/bacaro.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
#include <sys/epoll.h>
#include <cstdlib>
#include <filesystem>
#include <fstream>
#include <iomanip>
#include <random>
#include <sstream>
Expand Down Expand Up @@ -77,7 +78,18 @@ static void drain_zmq(void *sock, const std::function<void(void *)> &handler)

// ── Lifecycle ─────────────────────────────────────────────────────────────────

bacaro_t *bacaro_new(const char *name)
static void write_manifest(bacaro_t *self, const char **published_domains)
{
if (!published_domains)
return;

std::string path = ipc_path(self, ".manifest");
std::ofstream out(path);
for (const char **p = published_domains; *p; ++p)
out << *p << '\n';
}

bacaro_t *bacaro_new(const char *name, const char **published_domains)
{
if (!name || name[0] == '\0')
return nullptr;
Expand All @@ -96,6 +108,9 @@ bacaro_t *bacaro_new(const char *name)
if (ec)
goto fail;

// Write manifest before discovery so peers see it when they find our .pub
write_manifest(self, published_domains);

self->zmq_ctx = zmq_ctx_new();
if (!self->zmq_ctx)
goto fail;
Expand Down Expand Up @@ -155,6 +170,9 @@ void bacaro_destroy(bacaro_t **self_ptr)
close_and_remove(self->pub_sock, ".pub");
close_and_remove(self->router_sock, ".rep");

// Remove manifest file if it exists
{ std::error_code ec; fs::remove(ipc_path(self, ".manifest"), ec); }

if (self->zmq_ctx) {
zmq_ctx_destroy(self->zmq_ctx);
self->zmq_ctx = nullptr;
Expand All @@ -181,8 +199,11 @@ int bacaro_subscribe(bacaro_t *self, const char *domain)
// Apply to the shared SUB socket (covers all connected peers)
zmq_setsockopt(self->sub_sock, ZMQ_SUBSCRIBE, prefix.c_str(), prefix.size());

// Ensure each peer has a DEALER socket and request snapshot
// Ensure each peer has a DEALER socket and request snapshot.
// Skip peers whose manifest doesn't overlap with this subscription.
for (auto &[filename, peer] : self->peers) {
if (!manifest_overlaps(peer, prefix))
continue;
if (discovery_ensure_dealer(self, filename, peer) == BACARO_OK)
snapshot_send_request(peer.dealer_sock, prefix);
}
Expand Down
54 changes: 45 additions & 9 deletions src/discovery.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
#include <sys/inotify.h>
#include <unistd.h>
#include <filesystem>
#include <fstream>
#include <cstring>

namespace fs = std::filesystem;
Expand Down Expand Up @@ -73,6 +74,27 @@ void discovery_epoll_remove_zmq(bacaro_t *self, void *sock)
epoll_modify(self->epoll_fd, get_zmq_fd(sock), EPOLL_CTL_DEL);
}

static void read_manifest(const bacaro_t *self, const std::string &pub_filename,
std::vector<std::string> &out, bool &found)
{
// Replace .pub suffix with .manifest
std::string manifest_file = self->runtime_dir + "/"
+ pub_filename.substr(0, pub_filename.size() - 4) + ".manifest";

std::ifstream in(manifest_file);
if (!in.is_open()) {
found = false;
return;
}

found = true;
std::string line;
while (std::getline(in, line)) {
if (!line.empty())
out.push_back(std::move(line));
}
}

static std::string ipc_endpoint(const bacaro_t *self, const std::string &filename)
{
return "ipc://" + self->runtime_dir + "/" + filename;
Expand Down Expand Up @@ -103,18 +125,32 @@ int discovery_peer_connect(bacaro_t *self, const std::string &filename)
return BACARO_EZMQ;

std::string rep_ep = ipc_endpoint(self, rep_filename);
self->peers[filename] = { nullptr, peer_name, pub_ep, rep_ep, -1 };

// ── Lazy DEALER: only create if we have active subscriptions ─────────
PeerInfo peer_info;
peer_info.name = peer_name;
peer_info.pub_endpoint = pub_ep;
peer_info.rep_endpoint = rep_ep;

// Read manifest if available
read_manifest(self, filename, peer_info.manifest, peer_info.has_manifest);

self->peers[filename] = std::move(peer_info);

// ── Lazy DEALER: only create if subscriptions overlap with manifest ─
if (!self->subscriptions.empty()) {
if (discovery_ensure_dealer(self, filename, self->peers[filename]) != BACARO_OK) {
zmq_disconnect(self->sub_sock, pub_ep.c_str());
self->peers.erase(filename);
return BACARO_EZMQ;
auto &peer = self->peers[filename];
for (const auto &prefix : self->subscriptions) {
if (!manifest_overlaps(peer, prefix))
continue;
if (!peer.dealer_sock) {
if (discovery_ensure_dealer(self, filename, peer) != BACARO_OK) {
zmq_disconnect(self->sub_sock, pub_ep.c_str());
self->peers.erase(filename);
return BACARO_EZMQ;
}
}
snapshot_send_request(peer.dealer_sock, prefix);
}

for (const auto &prefix : self->subscriptions)
snapshot_send_request(self->peers[filename].dealer_sock, prefix);
}

return BACARO_OK;
Expand Down
19 changes: 19 additions & 0 deletions src/internal.h
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,27 @@ struct PeerInfo {
std::string pub_endpoint; // "ipc://..." for zmq_disconnect on the shared SUB
std::string rep_endpoint; // "ipc://..." for lazy DEALER connect
int dealer_fd = -1; // ZMQ_FD of dealer_sock (epoll)
std::vector<std::string> manifest; // declared published domains (empty if no manifest)
bool has_manifest = false; // true if peer had a .manifest file at discovery time
};

// Check if a subscription prefix overlaps with a peer's manifest.
// Returns true if the peer has no manifest or if any declared domain matches.
static inline bool manifest_overlaps(const PeerInfo &peer, const std::string &prefix)
{
if (!peer.has_manifest)
return true;
for (const auto &domain : peer.manifest) {
if (prefix.empty() || domain == prefix
|| (domain.size() > prefix.size() && domain[prefix.size()] == '.'
&& domain.compare(0, prefix.size(), prefix) == 0)
|| (prefix.size() > domain.size() && prefix[domain.size()] == '.'
&& prefix.compare(0, domain.size(), domain) == 0))
return true;
}
return false;
}

struct bacaro_s {
std::string name;
std::string runtime_dir;
Expand Down
Loading
Loading