Publish path, no dispatch -- FABRIC-3.6.md task 3.3

sk_hermes_publish() allocates one SkHermesMessage per channel member (heat-cost ruling 2026-09-21: one message per subscriber, funded by the publisher's own reservoir) and enqueues each onto a new per-subscriber SkHermesPendingQueue -- found-or-created lazily by vm_id, sized from stadium_max_vm_count() like the channel/switch tables. Best-effort across subscribers: a failed allocation or full queue skips and rolls back just that one subscriber, not the whole publish -- the natural reading of "ledger and stadium_conserved() hold across N publishes to M subscribers" (the task's own check), not a separate ruling. Dispatches nothing -- sk_hermes_pending_count()/peek()/pop() are the read/drain primitives task 3.4's real checkpoint-driven drain will build on; this task's own self-test uses them directly since no checkpoint hook exists yet. Verified live on all three architectures: pending-queue table sized 50/202/50 slots (tracking the channel table's own per-arch sizing), a synthetic publish self-test (2 publishes to 3 subscribers) confirms exact per-subscriber delivery counts, and ledger/stadium_conserved() invariants hold both mid-publish and after manually draining every queue back to baseline. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-22 00:49:22 -04:00
parent a4afdfa591
commit 1f6343bc03
7 changed files with 28105 additions and 1 deletions
@@ -392,4 +392,144 @@ int sk_hermes_channel_member_count(int channel_id) {
    return (int)sk_hermes_channels[channel_id].membership.count;
 }

+/* Task 3.3: per-subscriber pending queue table. See kernel_hermes.h's own
+ * doc comments on sk_hermes_queues_boot_init()/SK_HERMES_PENDING_MAX for
+ * scope and sizing reasoning. A queue is found-or-created lazily by
+ * vm_id, same shape as stadium.c's quota_slot_for_vm() and session.c's
+ * session_find() -- not pre-populated at birth like channel membership
+ * is, since not every VM ever receives a message. */
+typedef struct {
+    VMUuid vm_id;
+    int    in_use;
+    int    slots[SK_HERMES_PENDING_MAX]; /* sk_hermes_msgs[] indices, circular FIFO */
+    size_t head;
+    size_t count;
+} SkHermesPendingQueue;
+
+static SkHermesPendingQueue *sk_hermes_queues             = (SkHermesPendingQueue *)0;
+static int                   sk_hermes_queue_capacity_val = 0;
+
+int sk_hermes_queues_boot_init(void) {
+    size_t max_vm_count = stadium_max_vm_count();
+    SkHermesPendingQueue *table;
+    size_t i;
+
+    if (max_vm_count == 0) return -1; /* Stadium not yet initialized */
+
+    table = (SkHermesPendingQueue *)kmalloc(max_vm_count * sizeof(SkHermesPendingQueue));
+    if (!table) return -1;
+
+    for (i = 0; i < max_vm_count; i++) {
+        memset(&table[i], 0, sizeof(SkHermesPendingQueue));
+    }
+
+    sk_hermes_queues             = table;
+    sk_hermes_queue_capacity_val = (int)max_vm_count;
+
+    console_puts("Kernel-Hermes: ");
+    console_put_u64((uint64_t)max_vm_count);
+    console_println(" pending-queue slots");
+
+    return 0;
+}
+
+static SkHermesPendingQueue *find_queue(VMUuid vm_id) {
+    int i;
+    if (!sk_hermes_queues) return (SkHermesPendingQueue *)0;
+    for (i = 0; i < sk_hermes_queue_capacity_val; i++) {
+        if (sk_hermes_queues[i].in_use && vm_uuid_equal(sk_hermes_queues[i].vm_id, vm_id)) {
+            return &sk_hermes_queues[i];
+        }
+    }
+    return (SkHermesPendingQueue *)0;
+}
+
+static SkHermesPendingQueue *find_or_create_queue(VMUuid vm_id) {
+    int i, free_slot = -1;
+
+    if (!sk_hermes_queues) return (SkHermesPendingQueue *)0;
+
+    for (i = 0; i < sk_hermes_queue_capacity_val; i++) {
+        if (sk_hermes_queues[i].in_use && vm_uuid_equal(sk_hermes_queues[i].vm_id, vm_id)) {
+            return &sk_hermes_queues[i];
+        }
+        if (!sk_hermes_queues[i].in_use && free_slot < 0) free_slot = i;
+    }
+    if (free_slot < 0) return (SkHermesPendingQueue *)0; /* table full */
+
+    memset(&sk_hermes_queues[free_slot], 0, sizeof(SkHermesPendingQueue));
+    sk_hermes_queues[free_slot].vm_id  = vm_id;
+    sk_hermes_queues[free_slot].in_use = 1;
+    return &sk_hermes_queues[free_slot];
+}
+
+static int queue_push(SkHermesPendingQueue *q, int msg_index) {
+    size_t tail;
+    if (q->count >= SK_HERMES_PENDING_MAX) return -1;
+    tail = (q->head + q->count) % SK_HERMES_PENDING_MAX;
+    q->slots[tail] = msg_index;
+    q->count++;
+    return 0;
+}
+
+int sk_hermes_publish(VMUuid from, int channel_id, uint32_t type,
+                       void *payload_addr, uint32_t payload_len) {
+    SkHermesMembership *m;
+    size_t i;
+    int delivered = 0;
+
+    if (!channel_valid(channel_id)) return -1;
+
+    m = &sk_hermes_channels[channel_id].membership;
+    for (i = 0; i < m->count; i++) {
+        VMUuid to = m->members[i];
+        SkHermesMessage *msg;
+        SkHermesPendingQueue *q;
+        int idx;
+
+        if (sk_hermes_alloc(from, &msg) != 0) continue; /* this subscriber skipped, not fatal */
+
+        msg->type         = type;
+        msg->from         = from;
+        msg->to           = to;
+        msg->payload_addr = payload_addr;
+        msg->payload_len  = payload_len;
+        msg->channel      = (uint32_t)channel_id;
+
+        q = find_or_create_queue(to);
+        if (!q) {
+            sk_hermes_release(msg); /* roll back this subscriber's allocation */
+            continue;
+        }
+
+        idx = (int)(msg - sk_hermes_msgs);
+        if (queue_push(q, idx) != 0) {
+            sk_hermes_release(msg); /* this subscriber's own queue is full */
+            continue;
+        }
+
+        delivered++;
+    }
+    return delivered;
+}
+
+int sk_hermes_pending_count(VMUuid vm_id) {
+    SkHermesPendingQueue *q = find_queue(vm_id);
+    return q ? (int)q->count : 0;
+}
+
+SkHermesMessage *sk_hermes_pending_peek(VMUuid vm_id) {
+    SkHermesPendingQueue *q = find_queue(vm_id);
+    if (!q || q->count == 0) return (SkHermesMessage *)0;
+    return &sk_hermes_msgs[q->slots[q->head]];
+}
+
+int sk_hermes_pending_pop(VMUuid vm_id) {
+    SkHermesPendingQueue *q = find_queue(vm_id);
+    if (!q || q->count == 0) return -1;
+    q->head = (q->head + 1) % SK_HERMES_PENDING_MAX;
+    q->count--;
+    return 0;
+}
+
 #endif /* __STARKERNEL__ */