event_bus.hpp

Go to the documentation of this file.

// SPDX-License-Identifier: AGPL-3.0-or-later
// Copyright 2026 Johnson Ogundeji
//
// Routes SharedColumnBatch events through three tiers:
//
//   Tier 1 — SPSC dedicated channels (sub-μs, one per producer-consumer pair)
//             MpmcRing used as SPSC when only one writer; no locking overhead.
//             Use: exchange adapter → specific strategy engine.
//
//   Tier 2 — MPMC shared pool (μs, N producers → M worker threads)
//             All publish() calls land in a single MpmcRing shared by all
//             consumers.  Workers call pop() in a tight loop.
//             Use: all exchanges → ML inference pool (load-balanced).
//
//   Tier 3 — WAL / StreamingSink (ms, async durable logging)
//             Every publish() optionally serialises the batch and submits a
//             StreamRecord to an attached StreamingSink.  Non-blocking: drops
//             if the sink ring is full (increments tier3_drops counter).
//             Use: compliance log, Parquet compaction.
//
// Usage:
//   EventBus bus;
//
//   // Tier 1: dedicated channel
//   auto ch = bus.make_channel("binance→risk", 256);
//   ch->push(batch);        // producer (Binance adapter thread)
//   ch->pop(batch);         // consumer (risk gate thread)
//
//   // Tier 2: shared pool
//   bus.publish(batch);     // any producer
//   bus.pop(batch);         // any worker thread
//
//   // Tier 3: attach sink for compliance logging
//   bus.attach_sink(&my_streaming_sink);
//
// Phase 9b: MPMC ColumnBatch Event Bus.
 
#pragma once
 
#include "signet/error.hpp"
#include "signet/ai/mpmc_ring.hpp"
#include "signet/ai/column_batch.hpp"
#include "signet/ai/streaming_sink.hpp"
 
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>
 
namespace signet::forge {
 
// ============================================================================
// EventBusOptions — defined outside EventBus for Apple Clang compat
// ============================================================================
 
struct EventBusOptions {
    size_t tier2_capacity = 4096;   
    size_t tier1_capacity = 256;    
    bool   enable_tier3   = false;  
};
 
// ============================================================================
// EventBus — multi-tier SharedColumnBatch router
// ============================================================================
 
class EventBus {
public:
    using Options = EventBusOptions;
 
    // -------------------------------------------------------------------------
    // Channel — Tier-1 dedicated MpmcRing used as SPSC
    // -------------------------------------------------------------------------
 
    using Channel = MpmcRing<SharedColumnBatch>;
 
    // -------------------------------------------------------------------------
    // Construction
    // -------------------------------------------------------------------------
 
    explicit EventBus(Options opts = {})
        : opts_(opts)
        , tier2_(std::make_unique<MpmcRing<SharedColumnBatch>>(
                     opts.tier2_capacity)) {}
 
    EventBus(EventBus&&) = delete;            
    EventBus& operator=(EventBus&&) = delete; 
    EventBus(const EventBus&) = delete;
    EventBus& operator=(const EventBus&) = delete;
    ~EventBus() = default;
 
    // =========================================================================
    // Tier 1: dedicated per-channel rings
    // =========================================================================
 
    [[nodiscard]] std::shared_ptr<Channel> make_channel(
            const std::string& name,
            size_t capacity = 0) {
        std::lock_guard<std::mutex> lk(channels_mutex_);
        auto it = channels_.find(name);
        if (it != channels_.end()) return it->second;
 
        const size_t cap = (capacity == 0) ? opts_.tier1_capacity : capacity;
        auto ch = std::make_shared<Channel>(cap);
        channels_.emplace(name, ch);
        return ch;
    }
 
    [[nodiscard]] std::shared_ptr<Channel> channel(
            const std::string& name) const {
        std::lock_guard<std::mutex> lk(channels_mutex_);
        auto it = channels_.find(name);
        return (it != channels_.end()) ? it->second : nullptr;
    }
 
    // =========================================================================
    // Tier 2: shared MPMC pool
    // =========================================================================
 
    bool publish(SharedColumnBatch batch) {
        // Tier 3 first — atomic load ensures sink stays alive for the call
        // (lock-free: no mutex on the publish() hot path)
        auto sink = std::atomic_load_explicit(&sink_, std::memory_order_acquire);
        if (opts_.enable_tier3 && sink) {
            auto rec = batch->to_stream_record();
            auto r   = sink->submit(std::move(rec));
            if (!r) tier3_drops_.fetch_add(1, std::memory_order_relaxed);
        }
 
        // Tier 2
        if (!tier2_->push(std::move(batch))) {
            dropped_.fetch_add(1, std::memory_order_relaxed);
            return false;
        }
        published_.fetch_add(1, std::memory_order_relaxed);
        return true;
    }
 
    bool pop(SharedColumnBatch& out) {
        return tier2_->pop(out);
    }
 
    // =========================================================================
    // Tier 3: WAL / StreamingSink
    // =========================================================================
 
    void attach_sink(std::shared_ptr<StreamingSink> sink) {
        std::atomic_store_explicit(&sink_, std::move(sink),
                                   std::memory_order_release);
    }
 
    void detach_sink() {
        std::atomic_store_explicit(&sink_, std::shared_ptr<StreamingSink>{},
                                   std::memory_order_release);
    }
 
    [[nodiscard]] bool has_sink() const {
        return std::atomic_load_explicit(&sink_, std::memory_order_acquire)
                   != nullptr;
    }
 
    // =========================================================================
    // Stats
    // =========================================================================
 
    struct Stats {
        uint64_t published   = 0;   
        uint64_t dropped     = 0;   
        uint64_t tier3_drops = 0;   
    };
 
    [[nodiscard]] Stats stats() const noexcept {
        return Stats{
            published_.load(std::memory_order_relaxed),
            dropped_.load(std::memory_order_relaxed),
            tier3_drops_.load(std::memory_order_relaxed)
        };
    }
 
    void reset_stats() noexcept {
        published_.store(0, std::memory_order_relaxed);
        dropped_.store(0, std::memory_order_relaxed);
        tier3_drops_.store(0, std::memory_order_relaxed);
    }
 
    // =========================================================================
    // Introspection
    // =========================================================================
 
    [[nodiscard]] size_t tier2_size()     const noexcept { return tier2_->size(); }
    [[nodiscard]] size_t tier2_capacity() const noexcept { return tier2_->capacity(); }
    [[nodiscard]] size_t num_channels()   const {
        std::lock_guard<std::mutex> lk(channels_mutex_);
        return channels_.size();
    }
 
private:
    Options opts_;
 
    // Tier 2 — shared MPMC ring
    std::unique_ptr<MpmcRing<SharedColumnBatch>> tier2_;
 
    // Tier 1 — named dedicated channels
    mutable std::mutex channels_mutex_;
    std::unordered_map<std::string, std::shared_ptr<Channel>> channels_;
 
    // Tier 3 — WAL sink (shared ownership; lock-free via atomic shared_ptr).
    // CWE-362: H-16 fix — atomic_load/store on shared_ptr prevents
    // use-after-detach race without mutex overhead on the publish() hot path.
    std::shared_ptr<StreamingSink> sink_;
 
    // Stats
    std::atomic<uint64_t> published_{0};
    std::atomic<uint64_t> dropped_{0};
    std::atomic<uint64_t> tier3_drops_{0};
};
 
} // namespace signet::forge