Per-process manifest for targeted snapshot requests (#13)

Author: bricke

docs/api.md

@@ -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
 
@@ -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);
@@ -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);
 

docs/architecture.md

@@ -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.
 
 ---
@@ -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`)
 
@@ -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

docs/concepts.md

@@ -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.
 
@@ -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
 
@@ -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.

include/bacaro.h

@@ -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 */

llms.txt

@@ -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

src/bacaro.cpp

@@ -6,6 +6,7 @@
 #include <sys/epoll.h>
 #include <cstdlib>
 #include <filesystem>
+#include <fstream>
 #include <iomanip>
 #include <random>
 #include <sstream>
@@ -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;
@@ -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;
@@ -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;
@@ -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);
     }

src/discovery.cpp

@@ -5,6 +5,7 @@
 #include <sys/inotify.h>
 #include <unistd.h>
 #include <filesystem>
+#include <fstream>
 #include <cstring>
 
 namespace fs = std::filesystem;
@@ -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;
@@ -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;

src/internal.h

@@ -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;