From fd2d3de359c27bd9ae49a54adfc425457f02a211 Mon Sep 17 00:00:00 2001 From: torlando-tech Date: Tue, 5 May 2026 01:44:50 -0400 Subject: [PATCH] Track A: pyxis builds against attermann/microReticulum @ 0.3.0 + fixes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 🎯 First successful build against vanilla upstream (+ our PKCS7/HMAC/X25519 fixes branch). Flash 71.3% / RAM 21.4% on tdeck env. Final shim trim + API patches to get the build through: - Drop fork-only files from lib/microreticulum-shim/ that aren't referenced anywhere outside the shim itself: Buffer.cpp/h, ChannelData.h — fork's Channel work, unused by LXMF Cryptography/Ratchet.cpp/h — RNS 1.x compat, included `` directly; not used by anything pyxis touches today SegmentAccumulator.cpp/h — fork's Resource segmentation, depends on Resource methods vanilla doesn't expose; not used outside shim - LXMessage default-init Destination members with {RNS::Type::NONE} so the default LXMessage() ctor isn't implicitly deleted (vanilla Destination has no default ctor). - LXMF/LXMRouter::static_resource_concluded_callback stubbed to a no-op + ERROR log: vanilla Resource doesn't expose link(). Resource-form inbound LXMF delivery is currently disabled (PROPAGATION ⇄ LXMessage glue is broken until Resource API is reconciled). Tracked in spike doc. - Link::pending_requests_count() -> .pending_requests().size(). What works on this build: - Compile + link (firmware.elf produced, 2.24MB flash) - Crypto path goes through our pyxis-fixes-on-0.3.0 branch which has spec-correct PKCS7, HMAC, and X25519 clamping - microStore-based SPIFFS persistence via the SPIFFSFileSystem adapter - All BLE/SX1262/auto-interface adapters - LXMF outbound + inbound for non-Resource (DIRECT-via-Link, OPPORTUNISTIC) What's broken on this build (known, deliberate, tracked): - Resource-form inbound LXMF delivery (propagation node sync doesn't receive messages — outbound path still works for sending) - Transport memory diagnostics ([TABLES] dumps replaced with placeholder) - LoRa RSSI display + BLE peer-count display in the LXMF UI - Identity::mark_persistent (the 5s fast-flush no longer exists; fall back to microStore's dirty-tracking) Next: runtime smoke test on T-Deck (10.0.0.177 OTA) to validate that the above "works" surfaces actually work end-to-end against the python reference impl. Conformance bridge already passes 52/85 against this exact submodule pin. --- lib/microreticulum-shim/Buffer.cpp | 404 ------------------ lib/microreticulum-shim/Buffer.h | 163 ------- lib/microreticulum-shim/ChannelData.h | 369 ---------------- .../Cryptography/Ratchet.cpp | 193 --------- .../Cryptography/Ratchet.h | 181 -------- lib/microreticulum-shim/LXMF/LXMRouter.cpp | 35 +- lib/microreticulum-shim/LXMF/LXMessage.h | 9 +- .../SegmentAccumulator.cpp | 231 ---------- lib/microreticulum-shim/SegmentAccumulator.h | 162 ------- 9 files changed, 28 insertions(+), 1719 deletions(-) delete mode 100644 lib/microreticulum-shim/Buffer.cpp delete mode 100644 lib/microreticulum-shim/Buffer.h delete mode 100644 lib/microreticulum-shim/ChannelData.h delete mode 100644 lib/microreticulum-shim/Cryptography/Ratchet.cpp delete mode 100644 lib/microreticulum-shim/Cryptography/Ratchet.h delete mode 100644 lib/microreticulum-shim/SegmentAccumulator.cpp delete mode 100644 lib/microreticulum-shim/SegmentAccumulator.h diff --git a/lib/microreticulum-shim/Buffer.cpp b/lib/microreticulum-shim/Buffer.cpp deleted file mode 100644 index 78e26482..00000000 --- a/lib/microreticulum-shim/Buffer.cpp +++ /dev/null @@ -1,404 +0,0 @@ -#include "Buffer.h" -#include "ChannelData.h" // Required for Channel template instantiation -#include "Cryptography/BZ2.h" - -#include - -namespace RNS { - -//============================================================================== -// RawChannelReaderData - Internal data for RawChannelReader -//============================================================================== - -class RawChannelReaderData { -public: - RawChannelReaderData(uint16_t stream_id, Channel& channel) - : _stream_id(stream_id), _channel(&channel), _eof(false), _closed(false) { - MEM("RawChannelReaderData object created"); - } - - ~RawChannelReaderData() { - if (!_closed) { - close(); - } - MEM("RawChannelReaderData object destroyed"); - } - - bool handle_message(MessageBase& msg) { - if (_closed) return false; - - // Only handle StreamDataMessage - if (msg.msgtype() != StreamDataMessage::MSGTYPE) { - return false; - } - - StreamDataMessage& stream_msg = static_cast(msg); - - // Filter by stream_id - if (stream_msg.stream_id != _stream_id) { - return false; - } - - DEBUGF("RawChannelReader: Received %zu bytes on stream %u (eof=%d)", - stream_msg.data.size(), _stream_id, stream_msg.eof); - - // Accumulate data - if (stream_msg.data) { - _buffer += stream_msg.data; - } - - // Set EOF flag - if (stream_msg.eof) { - _eof = true; - DEBUG("RawChannelReader: EOF received"); - } - - // Notify callbacks - notify_ready(); - - return true; // Message consumed - } - - void notify_ready() { - size_t avail = _buffer.size(); - if (avail == 0 && !_eof) return; - - // Make a copy of callbacks in case they modify the list - auto callbacks_copy = _ready_callbacks; - for (auto& callback : callbacks_copy) { - try { - callback(avail); - } catch (...) { - ERROR("RawChannelReader: Callback threw exception"); - } - } - } - - void close() { - _closed = true; - _ready_callbacks.clear(); - DEBUG("RawChannelReader: Closed"); - } - - uint16_t _stream_id; - Channel* _channel; - Bytes _buffer; - bool _eof; - bool _closed; - std::vector _ready_callbacks; -}; - -//============================================================================== -// StreamDataMessage -//============================================================================== - -StreamDataMessage::StreamDataMessage(uint16_t stream_id, const Bytes& data, - bool eof, bool compressed) - : stream_id(stream_id), data(data), eof(eof), compressed(compressed) { -} - -Bytes StreamDataMessage::pack() const { - // Build 2-byte header: EOF(bit15) | COMP(bit14) | STREAM_ID(bits13-0) - uint16_t header = (stream_id & Type::Buffer::STREAM_ID_MASK); - if (eof) header |= Type::Buffer::FLAG_EOF; - if (compressed) header |= Type::Buffer::FLAG_COMPRESSED; - - Bytes result; - result.reserve(Type::Buffer::STREAM_OVERHEAD + data.size()); - - // Big-endian header (matching Python struct.pack(">H", header)) - result.append(static_cast((header >> 8) & 0xFF)); - result.append(static_cast(header & 0xFF)); - - // Data payload (already compressed if compressed flag is set) - if (data) { - result += data; - } - - return result; -} - -void StreamDataMessage::unpack(const Bytes& raw) { - if (raw.size() < Type::Buffer::STREAM_OVERHEAD) { - ERROR("StreamDataMessage::unpack: Data too short"); - return; - } - - // Read big-endian header - uint16_t header = (static_cast(raw[0]) << 8) | raw[1]; - - eof = (header & Type::Buffer::FLAG_EOF) != 0; - compressed = (header & Type::Buffer::FLAG_COMPRESSED) != 0; - stream_id = header & Type::Buffer::STREAM_ID_MASK; - - // Extract data payload - if (raw.size() > Type::Buffer::STREAM_OVERHEAD) { - data = raw.mid(Type::Buffer::STREAM_OVERHEAD); - - // Decompress if needed - if (compressed && data.size() > 0) { - data = Cryptography::bz2_decompress(data); - } - } else { - data = Bytes::NONE; - } -} - -//============================================================================== -// RawChannelReader - Uses shared_ptr to RawChannelReaderData -//============================================================================== - -RawChannelReader::RawChannelReader(uint16_t stream_id, Channel& channel) - : _object(std::make_shared(stream_id, channel)) { - - // Register StreamDataMessage type (as system message) - _object->_channel->register_message_type(true); - - // Add message handler - capture shared_ptr to keep data alive - // and use weak_ptr to avoid preventing destruction - std::weak_ptr weak_obj = _object; - _object->_channel->add_message_handler([weak_obj](MessageBase& msg) -> bool { - if (auto obj = weak_obj.lock()) { - return obj->handle_message(msg); - } - return false; // Object destroyed, don't handle - }); - - DEBUGF("RawChannelReader: Created for stream_id=%u", stream_id); -} - -RawChannelReader::~RawChannelReader() { - // shared_ptr handles cleanup automatically - MEM("RawChannelReader object destroyed"); -} - -Bytes RawChannelReader::read(size_t max_bytes) { - if (!_object || _object->_buffer.size() == 0) { - return Bytes::NONE; - } - - size_t to_read = (max_bytes == 0) ? _object->_buffer.size() - : std::min(max_bytes, _object->_buffer.size()); - Bytes result = _object->_buffer.left(to_read); - _object->_buffer = _object->_buffer.mid(to_read); - - DEBUGF("RawChannelReader: Read %zu bytes, %zu remaining", to_read, _object->_buffer.size()); - return result; -} - -Bytes RawChannelReader::readline() { - if (!_object) return Bytes::NONE; - - // Search for newline - for (size_t i = 0; i < _object->_buffer.size(); i++) { - if (_object->_buffer[i] == '\n') { - // Return including the newline - Bytes line = _object->_buffer.left(i + 1); - _object->_buffer = _object->_buffer.mid(i + 1); - return line; - } - } - - // No complete line available - // If EOF, return remaining data as final line (no newline) - if (_object->_eof && _object->_buffer.size() > 0) { - Bytes line = _object->_buffer; - _object->_buffer = Bytes::NONE; - return line; - } - - return Bytes::NONE; -} - -size_t RawChannelReader::available() const { - if (!_object) return 0; - return _object->_buffer.size(); -} - -bool RawChannelReader::eof() const { - if (!_object) return true; - return _object->_eof && _object->_buffer.size() == 0; -} - -void RawChannelReader::add_ready_callback(ReadyCallback callback) { - if (!_object) return; - _object->_ready_callbacks.push_back(callback); -} - -void RawChannelReader::remove_ready_callback(ReadyCallback callback) { - // Note: Function comparison is tricky in C++ - // This is a simplified implementation - // For production, consider using callback IDs -} - -void RawChannelReader::close() { - if (!_object) return; - _object->close(); -} - -//============================================================================== -// RawChannelWriter -//============================================================================== - -RawChannelWriter::RawChannelWriter(uint16_t stream_id, Channel& channel) - : _stream_id(stream_id), _channel(&channel), _eof_sent(false) { - - // Calculate max data length (Channel MDU minus stream header) - _max_data_len = _channel->mdu() - Type::Buffer::STREAM_OVERHEAD; - - // Register StreamDataMessage type (as system message) - // This may already be registered by a reader, but that's OK - _channel->register_message_type(true); - - DEBUGF("RawChannelWriter: Created for stream_id=%u, max_data_len=%zu", - _stream_id, _max_data_len); -} - -RawChannelWriter::~RawChannelWriter() { - if (!_eof_sent && _channel) { - close(); - } -} - -RawChannelWriter::RawChannelWriter(RawChannelWriter&& other) noexcept - : _stream_id(other._stream_id), - _channel(other._channel), - _max_data_len(other._max_data_len), - _eof_sent(other._eof_sent) { - other._channel = nullptr; - other._eof_sent = true; -} - -RawChannelWriter& RawChannelWriter::operator=(RawChannelWriter&& other) noexcept { - if (this != &other) { - if (!_eof_sent && _channel) { - close(); - } - _stream_id = other._stream_id; - _channel = other._channel; - _max_data_len = other._max_data_len; - _eof_sent = other._eof_sent; - other._channel = nullptr; - other._eof_sent = true; - } - return *this; -} - -size_t RawChannelWriter::write(const Bytes& data) { - if (!_channel || _eof_sent) { - ERROR("RawChannelWriter: Cannot write after close"); - return 0; - } - - if (!data || data.size() == 0) { - return 0; - } - - size_t chunk_len = std::min(data.size(), Type::Buffer::MAX_CHUNK_LEN); - Bytes chunk = data.left(chunk_len); - - bool use_compression = false; - Bytes send_data; - size_t processed = 0; - - // Try compression at decreasing chunk sizes (matching Python Buffer.py) - if (chunk_len > Type::Buffer::COMPRESSION_MIN_SIZE) { - for (size_t try_num = 1; try_num <= Type::Buffer::COMPRESSION_TRIES; try_num++) { - size_t segment_len = chunk_len / try_num; - if (segment_len == 0) break; - - Bytes segment = chunk.left(segment_len); - Bytes compressed = Cryptography::bz2_compress(segment); - - if (compressed.size() < _max_data_len && - compressed.size() < segment_len) { - use_compression = true; - send_data = compressed; - processed = segment_len; - DEBUGF("RawChannelWriter: Compression succeeded: %zu -> %zu bytes", - segment_len, compressed.size()); - break; - } - } - } - - // If compression didn't help, send uncompressed - if (!use_compression) { - send_data = chunk.left(_max_data_len); - processed = send_data.size(); - } - - // Build and send message - StreamDataMessage msg; - msg.stream_id = _stream_id; - msg.data = send_data; - msg.eof = false; - msg.compressed = use_compression; - - _channel->send(msg); - - DEBUGF("RawChannelWriter: Sent %zu bytes on stream %u (compressed=%d)", - processed, _stream_id, use_compression); - - return processed; -} - -void RawChannelWriter::flush() { - // In the current implementation, write() sends immediately - // This is a no-op but provided for API compatibility - DEBUG("RawChannelWriter: Flush called (no-op)"); -} - -void RawChannelWriter::close() { - if (_eof_sent) return; - - if (_channel) { - // Send EOF message - StreamDataMessage msg; - msg.stream_id = _stream_id; - msg.data = Bytes::NONE; - msg.eof = true; - msg.compressed = false; - - _channel->send(msg); - - DEBUGF("RawChannelWriter: Sent EOF on stream %u", _stream_id); - } - - _eof_sent = true; -} - -//============================================================================== -// Buffer namespace factory functions -//============================================================================== - -namespace Buffer { - -RawChannelReader create_reader(uint16_t stream_id, Channel& channel, - RawChannelReader::ReadyCallback callback) { - RawChannelReader reader(stream_id, channel); - if (callback) { - reader.add_ready_callback(callback); - } - return reader; -} - -RawChannelWriter create_writer(uint16_t stream_id, Channel& channel) { - return RawChannelWriter(stream_id, channel); -} - -std::pair -create_bidirectional_buffer(uint16_t rx_stream_id, uint16_t tx_stream_id, - Channel& channel, - RawChannelReader::ReadyCallback callback) { - RawChannelReader reader(rx_stream_id, channel); - if (callback) { - reader.add_ready_callback(callback); - } - RawChannelWriter writer(tx_stream_id, channel); - return std::make_pair(std::move(reader), std::move(writer)); -} - -} // namespace Buffer - -} // namespace RNS diff --git a/lib/microreticulum-shim/Buffer.h b/lib/microreticulum-shim/Buffer.h deleted file mode 100644 index b5016c35..00000000 --- a/lib/microreticulum-shim/Buffer.h +++ /dev/null @@ -1,163 +0,0 @@ -#pragma once - -#ifndef RNS_BUFFER_H -#define RNS_BUFFER_H - -#include "Bytes.h" -#include "Channel.h" -#include "MessageBase.h" -#include "Type.h" -#include "Log.h" - -#include -#include -#include -#include - -namespace RNS { - -/** - * StreamDataMessage - Wire format for Buffer data over Channel - * - * Header (2 bytes, big-endian): - * Bit 15: EOF flag - * Bit 14: Compression flag - * Bits 13-0: Stream ID (max 16383) - * - * Followed by optional data payload (BZ2 compressed if flag set) - */ -class StreamDataMessage : public MessageBase { -public: - static constexpr uint16_t MSGTYPE = Type::Channel::SMT_STREAM_DATA; - - // Fields - uint16_t stream_id = 0; - Bytes data; - bool eof = false; - bool compressed = false; - - // Constructors - StreamDataMessage() = default; - StreamDataMessage(uint16_t stream_id, const Bytes& data = Bytes::NONE, - bool eof = false, bool compressed = false); - - // MessageBase interface - uint16_t msgtype() const override { return MSGTYPE; } - Bytes pack() const override; - void unpack(const Bytes& raw) override; -}; - -// Forward declaration for internal data -class RawChannelReaderData; - -/** - * RawChannelReader - Read stream data from a Channel - * - * Uses shared_ptr internally so that move operations preserve callback validity. - */ -class RawChannelReader { -public: - using ReadyCallback = std::function; - - RawChannelReader(uint16_t stream_id, Channel& channel); - RawChannelReader(Type::NoneConstructor none) { MEM("RawChannelReader NONE object created"); } - ~RawChannelReader(); - - // Copy/move - shallow copy of shared_ptr (like other RNS types) - RawChannelReader(const RawChannelReader& other) : _object(other._object) { - MEM("RawChannelReader object copy created"); - } - RawChannelReader(RawChannelReader&& other) noexcept : _object(std::move(other._object)) { - MEM("RawChannelReader object moved"); - } - RawChannelReader& operator=(const RawChannelReader& other) { - _object = other._object; - return *this; - } - RawChannelReader& operator=(RawChannelReader&& other) noexcept { - _object = std::move(other._object); - return *this; - } - - // Validity check - operator bool() const { return _object.get() != nullptr; } - - // Reading interface - Bytes read(size_t max_bytes = 0); // 0 = read all available - Bytes readline(); - size_t available() const; - bool eof() const; - - // Callback management - void add_ready_callback(ReadyCallback callback); - void remove_ready_callback(ReadyCallback callback); - - // Cleanup - void close(); - -private: - std::shared_ptr _object; -}; - -/** - * RawChannelWriter - Write stream data to a Channel - */ -class RawChannelWriter { -public: - RawChannelWriter(uint16_t stream_id, Channel& channel); - ~RawChannelWriter(); - - // Disable copy - RawChannelWriter(const RawChannelWriter&) = delete; - RawChannelWriter& operator=(const RawChannelWriter&) = delete; - - // Move support - RawChannelWriter(RawChannelWriter&& other) noexcept; - RawChannelWriter& operator=(RawChannelWriter&& other) noexcept; - - // Writing interface - size_t write(const Bytes& data); - void flush(); - void close(); // Send EOF - -private: - uint16_t _stream_id; - Channel* _channel; - size_t _max_data_len; - bool _eof_sent = false; -}; - -/** - * Buffer namespace - Factory functions for creating readers/writers - */ -namespace Buffer { - - /** - * Create a reader for receiving stream data - */ - RawChannelReader create_reader(uint16_t stream_id, Channel& channel, - RawChannelReader::ReadyCallback callback = nullptr); - - /** - * Create a writer for sending stream data - */ - RawChannelWriter create_writer(uint16_t stream_id, Channel& channel); - - /** - * Create bidirectional buffer pair - * @param rx_stream_id Stream ID for receiving data - * @param tx_stream_id Stream ID for sending data - * @param channel The Channel to use - * @param callback Optional callback when data is ready to read - * @return Pair of (reader, writer) - */ - std::pair - create_bidirectional_buffer(uint16_t rx_stream_id, uint16_t tx_stream_id, - Channel& channel, - RawChannelReader::ReadyCallback callback = nullptr); - -} // namespace Buffer - -} // namespace RNS - -#endif // RNS_BUFFER_H diff --git a/lib/microreticulum-shim/ChannelData.h b/lib/microreticulum-shim/ChannelData.h deleted file mode 100644 index c6a9170e..00000000 --- a/lib/microreticulum-shim/ChannelData.h +++ /dev/null @@ -1,369 +0,0 @@ -#pragma once - -#ifndef RNS_CHANNEL_DATA_H -#define RNS_CHANNEL_DATA_H - -#include "Bytes.h" -#include "Link.h" -#include "Packet.h" -#include "MessageBase.h" -#include "Type.h" -#include "Log.h" - -#include -#include -#include -#include -#include // for std::move - -namespace RNS { - -// Forward declaration -class Channel; - -// Envelope wraps a message with protocol metadata -class Envelope { -public: - Envelope() = default; - Envelope(uint16_t msgtype, uint16_t sequence, const Bytes& raw) - : _msgtype(msgtype), _sequence(sequence), _raw(raw) {} - - // Move semantics (unique_ptr makes class non-copyable) - Envelope(Envelope&&) = default; - Envelope& operator=(Envelope&&) = default; - - // Delete copy operations - Envelope(const Envelope&) = delete; - Envelope& operator=(const Envelope&) = delete; - - uint16_t msgtype() const { return _msgtype; } - uint16_t sequence() const { return _sequence; } - const Bytes& raw() const { return _raw; } - - // For TX tracking - Packet packet() const { return _packet; } - void set_packet(const Packet& packet) { _packet = packet; } - uint8_t tries() const { return _tries; } - void increment_tries() { _tries++; } - double timestamp() const { return _timestamp; } - void set_timestamp(double ts) { _timestamp = ts; } - bool tracked() const { return _tracked; } - void set_tracked(bool tracked) { _tracked = tracked; } - - // Message instance (for RX) - std::unique_ptr& message() { return _message; } - void set_message(std::unique_ptr msg) { _message = std::move(msg); } - - // Pack envelope to wire format (big-endian) - Bytes pack() const { - // Wire format: MSGTYPE(2) + SEQUENCE(2) + LENGTH(2) + DATA(N) - // All values big-endian - Bytes result; - - // Allocate exact size needed - size_t data_len = _raw.size(); - result.reserve(6 + data_len); - - // MSGTYPE (2 bytes, big-endian) - result.append((uint8_t)((_msgtype >> 8) & 0xFF)); - result.append((uint8_t)(_msgtype & 0xFF)); - - // SEQUENCE (2 bytes, big-endian) - result.append((uint8_t)((_sequence >> 8) & 0xFF)); - result.append((uint8_t)(_sequence & 0xFF)); - - // LENGTH (2 bytes, big-endian) - result.append((uint8_t)((data_len >> 8) & 0xFF)); - result.append((uint8_t)(data_len & 0xFF)); - - // DATA - result += _raw; - - return result; - } - - // Unpack envelope from wire format (big-endian) - static bool unpack(const Bytes& wire_data, Envelope& out) { - // Need at least 6 bytes for header - if (wire_data.size() < 6) { - return false; - } - - const uint8_t* data = wire_data.data(); - - // MSGTYPE (2 bytes, big-endian) - uint16_t msgtype = (static_cast(data[0]) << 8) | data[1]; - - // SEQUENCE (2 bytes, big-endian) - uint16_t sequence = (static_cast(data[2]) << 8) | data[3]; - - // LENGTH (2 bytes, big-endian) - uint16_t length = (static_cast(data[4]) << 8) | data[5]; - - // Validate length - if (wire_data.size() < 6 + static_cast(length)) { - return false; - } - - // Extract data payload - Bytes raw; - if (length > 0) { - raw = wire_data.mid(6, length); - } - - out = Envelope(msgtype, sequence, raw); - return true; - } - -private: - uint16_t _msgtype = 0; - uint16_t _sequence = 0; - Bytes _raw; - Packet _packet = {Type::NONE}; - uint8_t _tries = 0; - double _timestamp = 0.0; - bool _tracked = false; - std::unique_ptr _message; -}; - -// Internal Channel data structure -class ChannelData { -public: - enum class WindowTier { FAST, MEDIUM, SLOW, VERY_SLOW }; - - // Fixed ring buffer sizes - static constexpr size_t RX_RING_SIZE = 16; - static constexpr size_t TX_RING_SIZE = 16; - - ChannelData() { MEM("ChannelData object created"); } - ChannelData(const Link& link) : _link(link) { MEM("ChannelData object created with link"); } - virtual ~ChannelData() { MEM("ChannelData object destroyed"); } - - // RX Ring buffer operations (ordered by sequence) - bool rx_ring_empty() const { return _rx_ring_count == 0; } - size_t rx_ring_size() const { return _rx_ring_count; } - bool rx_ring_full() const { return _rx_ring_count >= RX_RING_SIZE; } - - Envelope& rx_ring_front() { - return _rx_ring_pool[_rx_ring_head]; - } - - void rx_ring_pop_front() { - if (_rx_ring_count > 0) { - // Reset the envelope being removed - _rx_ring_pool[_rx_ring_head] = Envelope(); - _rx_ring_head = (_rx_ring_head + 1) % RX_RING_SIZE; - _rx_ring_count--; - } - } - - void rx_ring_clear() { - // Reset all envelopes - for (size_t i = 0; i < RX_RING_SIZE; i++) { - _rx_ring_pool[i] = Envelope(); - } - _rx_ring_head = 0; - _rx_ring_tail = 0; - _rx_ring_count = 0; - } - - // Insert envelope in sequence order (for reordering) - // Returns false if ring is full - bool rx_ring_insert_ordered(Envelope&& envelope) { - if (_rx_ring_count >= RX_RING_SIZE) { - return false; - } - - uint16_t new_seq = envelope.sequence(); - - // If empty, just add at head - if (_rx_ring_count == 0) { - _rx_ring_pool[_rx_ring_head] = std::move(envelope); - _rx_ring_tail = (_rx_ring_head + 1) % RX_RING_SIZE; - _rx_ring_count = 1; - return true; - } - - // Find insertion position by iterating through valid entries - // We need to shift elements to make room - size_t insert_pos = _rx_ring_count; // Default: insert at end - - for (size_t i = 0; i < _rx_ring_count; i++) { - size_t idx = (_rx_ring_head + i) % RX_RING_SIZE; - uint16_t existing_seq = _rx_ring_pool[idx].sequence(); - - // Calculate relative position (handling wraparound) - int32_t diff = static_cast(new_seq) - static_cast(existing_seq); - if (diff >= static_cast(Type::Channel::SEQ_MODULUS / 2)) { - diff -= Type::Channel::SEQ_MODULUS; - } else if (diff < -static_cast(Type::Channel::SEQ_MODULUS / 2)) { - diff += Type::Channel::SEQ_MODULUS; - } - - if (diff < 0) { - // Insert before this position - insert_pos = i; - break; - } - } - - // Shift elements from insert_pos to end to make room - // We shift by moving tail back and shifting elements - for (size_t i = _rx_ring_count; i > insert_pos; i--) { - size_t dst_idx = (_rx_ring_head + i) % RX_RING_SIZE; - size_t src_idx = (_rx_ring_head + i - 1) % RX_RING_SIZE; - _rx_ring_pool[dst_idx] = std::move(_rx_ring_pool[src_idx]); - } - - // Insert the new envelope - size_t actual_idx = (_rx_ring_head + insert_pos) % RX_RING_SIZE; - _rx_ring_pool[actual_idx] = std::move(envelope); - _rx_ring_tail = (_rx_ring_tail + 1) % RX_RING_SIZE; - _rx_ring_count++; - - return true; - } - - // Check if sequence exists in RX ring - bool rx_ring_contains_sequence(uint16_t sequence) const { - for (size_t i = 0; i < _rx_ring_count; i++) { - size_t idx = (_rx_ring_head + i) % RX_RING_SIZE; - if (_rx_ring_pool[idx].sequence() == sequence) { - return true; - } - } - return false; - } - - // TX Ring buffer operations (simple FIFO with removal by packet match) - bool tx_ring_empty() const { return _tx_ring_count == 0; } - size_t tx_ring_size() const { return _tx_ring_count; } - bool tx_ring_full() const { return _tx_ring_count >= TX_RING_SIZE; } - - bool tx_ring_push_back(Envelope&& envelope) { - if (_tx_ring_count >= TX_RING_SIZE) { - return false; - } - _tx_ring_pool[_tx_ring_tail] = std::move(envelope); - _tx_ring_tail = (_tx_ring_tail + 1) % TX_RING_SIZE; - _tx_ring_count++; - return true; - } - - void tx_ring_clear() { - for (size_t i = 0; i < TX_RING_SIZE; i++) { - _tx_ring_pool[i] = Envelope(); - } - _tx_ring_head = 0; - _tx_ring_tail = 0; - _tx_ring_count = 0; - } - - // Remove envelope matching packet, returns true if found and removed - bool tx_ring_remove_by_packet(const Packet& packet) { - for (size_t i = 0; i < _tx_ring_count; i++) { - size_t idx = (_tx_ring_head + i) % TX_RING_SIZE; - if (_tx_ring_pool[idx].packet() == packet) { - // Shift remaining elements forward - for (size_t j = i; j < _tx_ring_count - 1; j++) { - size_t dst_idx = (_tx_ring_head + j) % TX_RING_SIZE; - size_t src_idx = (_tx_ring_head + j + 1) % TX_RING_SIZE; - _tx_ring_pool[dst_idx] = std::move(_tx_ring_pool[src_idx]); - } - // Clear the last slot - size_t last_idx = (_tx_ring_head + _tx_ring_count - 1) % TX_RING_SIZE; - _tx_ring_pool[last_idx] = Envelope(); - _tx_ring_count--; - return true; - } - } - return false; - } - - // Find envelope by packet (returns nullptr if not found) - Envelope* tx_ring_find_by_packet(const Packet& packet) { - for (size_t i = 0; i < _tx_ring_count; i++) { - size_t idx = (_tx_ring_head + i) % TX_RING_SIZE; - if (_tx_ring_pool[idx].packet() == packet) { - return &_tx_ring_pool[idx]; - } - } - return nullptr; - } - - // Find envelope by sequence (returns nullptr if not found) - Envelope* tx_ring_find_by_sequence(uint16_t sequence) { - for (size_t i = 0; i < _tx_ring_count; i++) { - size_t idx = (_tx_ring_head + i) % TX_RING_SIZE; - if (_tx_ring_pool[idx].sequence() == sequence) { - return &_tx_ring_pool[idx]; - } - } - return nullptr; - } - - // Iterate over TX ring (for counting outstanding, checking timeouts) - template - void tx_ring_foreach(Func&& func) { - for (size_t i = 0; i < _tx_ring_count; i++) { - size_t idx = (_tx_ring_head + i) % TX_RING_SIZE; - func(_tx_ring_pool[idx]); - } - } - - template - void tx_ring_foreach(Func&& func) const { - for (size_t i = 0; i < _tx_ring_count; i++) { - size_t idx = (_tx_ring_head + i) % TX_RING_SIZE; - func(_tx_ring_pool[idx]); - } - } - -private: - friend class Channel; - - // Link reference - Link _link = {Type::NONE}; - - // Sequencing - uint16_t _next_sequence = 0; - uint16_t _next_rx_sequence = 0; - - // RX Ring buffer (fixed-size circular buffer, ordered by sequence) - Envelope _rx_ring_pool[RX_RING_SIZE]; - size_t _rx_ring_head = 0; - size_t _rx_ring_tail = 0; - size_t _rx_ring_count = 0; - - // TX Ring buffer (fixed-size circular buffer) - Envelope _tx_ring_pool[TX_RING_SIZE]; - size_t _tx_ring_head = 0; - size_t _tx_ring_tail = 0; - size_t _tx_ring_count = 0; - - // Message dispatch - // Factory: msgtype -> function that creates a new message instance - // Note: These are only set up once at initialization, so map is acceptable - std::map()>> _message_factories; - // Handlers: list of callbacks, first returning true stops dispatch - // Note: These are only set up once at initialization, so vector is acceptable - std::vector> _message_callbacks; - - // Window management - uint16_t _window = Type::Channel::WINDOW_INITIAL; - uint16_t _window_min = Type::Channel::WINDOW_MIN; - uint16_t _window_max = Type::Channel::WINDOW_MAX; - uint16_t _fast_rate_rounds = 0; - - // Timing/RTT - double _rtt = 0.0; - uint8_t _max_tries = Type::Channel::MAX_TRIES; - WindowTier _current_tier = WindowTier::MEDIUM; - - // State - bool _ready = false; -}; - -} // namespace RNS - -#endif // RNS_CHANNEL_DATA_H diff --git a/lib/microreticulum-shim/Cryptography/Ratchet.cpp b/lib/microreticulum-shim/Cryptography/Ratchet.cpp deleted file mode 100644 index fed34215..00000000 --- a/lib/microreticulum-shim/Cryptography/Ratchet.cpp +++ /dev/null @@ -1,193 +0,0 @@ -#include "Ratchet.h" -#include "Fernet.h" -#include - -#include -#include - -#include -#include - -using namespace RNS; -using namespace RNS::Cryptography; - -// Constructor from existing key material -Ratchet::Ratchet(const Bytes& private_key, const Bytes& public_key, double created_at) - : _private_key(private_key), _public_key(public_key), _created_at(created_at) -{ - if (private_key.size() != RATCHET_LENGTH) { - throw std::invalid_argument("Ratchet private key must be exactly 32 bytes"); - } - if (public_key.size() != RATCHET_LENGTH) { - throw std::invalid_argument("Ratchet public key must be exactly 32 bytes"); - } - if (_created_at == 0.0) { - _created_at = Utilities::OS::time(); - } -} - -// Copy constructor -Ratchet::Ratchet(const Ratchet& ratchet) - : _private_key(ratchet._private_key) - , _public_key(ratchet._public_key) - , _created_at(ratchet._created_at) -{ -} - -// Assignment operator -Ratchet& Ratchet::operator=(const Ratchet& ratchet) { - if (this != &ratchet) { - _private_key = ratchet._private_key; - _public_key = ratchet._public_key; - _created_at = ratchet._created_at; - } - return *this; -} - -// Generate a new ratchet with fresh X25519 keypair -Ratchet Ratchet::generate() { - Bytes private_key; - Bytes public_key; - - // Generate random X25519 keypair using Curve25519::dh1 - // This matches X25519PrivateKey::generate() pattern - Curve25519::dh1(public_key.writable(RATCHET_LENGTH), private_key.writable(RATCHET_LENGTH)); - - double created_at = Utilities::OS::time(); - - TRACE("Ratchet::generate: Generated new ratchet"); - DEBUG(" Private key: " + private_key.toHex()); - DEBUG(" Public key: " + public_key.toHex()); - DEBUG(" Created at: " + std::to_string(created_at)); - - return Ratchet(private_key, public_key, created_at); -} - -// Derive ratchet ID from public key bytes -Bytes Ratchet::get_ratchet_id(const Bytes& public_bytes) { - if (public_bytes.size() != RATCHET_LENGTH) { - throw std::invalid_argument("Ratchet public key must be exactly 32 bytes"); - } - - // Ratchet ID is first 10 bytes of SHA-256(public_key) - // This matches Python implementation - Bytes hash = sha256(public_bytes); - Bytes ratchet_id = hash.left(RATCHET_ID_LENGTH); - - DEBUG("Ratchet::get_ratchet_id: " + ratchet_id.toHex() + " from pubkey " + public_bytes.toHex()); - - return ratchet_id; -} - -// Derive ratchet ID from X25519PublicKey object -Bytes Ratchet::get_ratchet_id(X25519PublicKey& public_key) { - return get_ratchet_id(public_key.public_bytes()); -} - -// Get public key bytes for announcement -Bytes Ratchet::public_bytes() const { - return _public_key; -} - -// Get private key bytes (use with caution!) -Bytes Ratchet::private_bytes() const { - return _private_key; -} - -// Get ratchet ID for this ratchet -Bytes Ratchet::get_id() const { - return get_ratchet_id(_public_key); -} - -// Derive shared secret with peer's ratchet -Bytes Ratchet::derive_shared_secret(const Bytes& peer_public_key) const { - if (!_private_key || !_public_key) { - throw std::runtime_error("Cannot derive shared secret from empty ratchet"); - } - if (peer_public_key.size() != RATCHET_LENGTH) { - throw std::invalid_argument("Peer public key must be exactly 32 bytes"); - } - - // Perform X25519 ECDH: shared = ECDH(my_private, peer_public) - // This uses Curve25519::eval() like X25519PrivateKey::exchange() - Bytes shared_secret; - if (!Curve25519::eval(shared_secret.writable(RATCHET_LENGTH), _private_key.data(), peer_public_key.data())) { - throw std::runtime_error("Peer ratchet key is invalid"); - } - - DEBUG("Ratchet::derive_shared_secret:"); - DEBUG(" My private: " + _private_key.toHex()); - DEBUG(" My public: " + _public_key.toHex()); - DEBUG(" Peer public: " + peer_public_key.toHex()); - DEBUG(" Shared: " + shared_secret.toHex()); - - return shared_secret; -} - -// Derive encryption/decryption key from shared secret -Bytes Ratchet::derive_key(const Bytes& shared_secret) const { - if (shared_secret.size() != RATCHET_LENGTH) { - throw std::invalid_argument("Shared secret must be exactly 32 bytes"); - } - - // Use HKDF to derive a Fernet-compatible key from the shared secret - // Fernet requires a 32-byte key (256 bits) - // Python implementation uses HKDF with no salt or context for ratchet keys - Bytes derived_key = hkdf(32, shared_secret); - - DEBUG("Ratchet::derive_key:"); - DEBUG(" Shared secret: " + shared_secret.toHex()); - DEBUG(" Derived key: " + derived_key.toHex()); - - return derived_key; -} - -// Encrypt plaintext using this ratchet and peer's public key -Bytes Ratchet::encrypt(const Bytes& plaintext, const Bytes& peer_public_key) const { - if (!_private_key || !_public_key) { - throw std::runtime_error("Cannot encrypt with empty ratchet"); - } - - // 1. Perform X25519 ECDH to get shared secret - Bytes shared_secret = derive_shared_secret(peer_public_key); - - // 2. Derive encryption key using HKDF - Bytes encryption_key = derive_key(shared_secret); - - // 3. Encrypt with Fernet - Fernet fernet(encryption_key); - Bytes ciphertext = fernet.encrypt(plaintext); - - DEBUG("Ratchet::encrypt: Encrypted " + std::to_string(plaintext.size()) + - " bytes to " + std::to_string(ciphertext.size()) + " bytes"); - - return ciphertext; -} - -// Decrypt ciphertext using this ratchet and peer's public key -Bytes Ratchet::decrypt(const Bytes& ciphertext, const Bytes& peer_public_key) const { - if (!_private_key || !_public_key) { - throw std::runtime_error("Cannot decrypt with empty ratchet"); - } - - try { - // 1. Perform X25519 ECDH to get shared secret - Bytes shared_secret = derive_shared_secret(peer_public_key); - - // 2. Derive decryption key using HKDF - Bytes decryption_key = derive_key(shared_secret); - - // 3. Decrypt with Fernet - Fernet fernet(decryption_key); - Bytes plaintext = fernet.decrypt(ciphertext); - - DEBUG("Ratchet::decrypt: Decrypted " + std::to_string(ciphertext.size()) + - " bytes to " + std::to_string(plaintext.size()) + " bytes"); - - return plaintext; - } - catch (const std::exception& e) { - ERROR("Ratchet::decrypt failed: " + std::string(e.what())); - throw; - } -} diff --git a/lib/microreticulum-shim/Cryptography/Ratchet.h b/lib/microreticulum-shim/Cryptography/Ratchet.h deleted file mode 100644 index e3f89cca..00000000 --- a/lib/microreticulum-shim/Cryptography/Ratchet.h +++ /dev/null @@ -1,181 +0,0 @@ -#pragma once - -#include -#include -#include -#include "X25519.h" -#include "HKDF.h" -#include "Fernet.h" -#include "Hashes.h" - -#include -#include - -namespace RNS { namespace Cryptography { - - /** - * @brief Represents a single ratchet (X25519 keypair) used for forward secrecy - * - * Ratchets provide forward secrecy by rotating encryption keys at regular intervals. - * Each ratchet consists of an X25519 keypair. The public key is announced, while - * the private key is used to derive shared secrets with peer ratchets. - */ - class Ratchet { - - public: - using Ptr = std::shared_ptr; - - // Constants from Python implementation - static const uint8_t RATCHET_LENGTH = 32; // X25519 key length - static const uint8_t RATCHET_ID_LENGTH = 10; // Truncated hash length - static const uint16_t MAX_RATCHETS = 128; // Maximum ratchets to store - static const uint32_t DEFAULT_RATCHET_INTERVAL = 1800; // 30 minutes in seconds - - public: - /** - * @brief Default constructor creates an empty ratchet - */ - Ratchet() : _created_at(0.0) {} - - /** - * @brief Constructs a ratchet from existing key material - * @param private_key The X25519 private key (32 bytes) - * @param public_key The X25519 public key (32 bytes) - * @param created_at Timestamp when ratchet was created - */ - Ratchet(const Bytes& private_key, const Bytes& public_key, double created_at = 0.0); - - /** - * @brief Copy constructor - */ - Ratchet(const Ratchet& ratchet); - - /** - * @brief Assignment operator - */ - Ratchet& operator=(const Ratchet& ratchet); - - /** - * @brief Destructor - */ - ~Ratchet() {} - - /** - * @brief Validity check - */ - inline operator bool() const { - return _private_key && _public_key; - } - - public: - /** - * @brief Generates a new ratchet with fresh X25519 keypair - * @return A new Ratchet instance - */ - static Ratchet generate(); - - /** - * @brief Derives a ratchet ID from public key bytes - * - * The ratchet ID is used to identify which ratchet was used to encrypt a packet. - * It's computed as the first 10 bytes of SHA-256(public_key). - * - * @param public_bytes The X25519 public key (32 bytes) - * @return Ratchet ID (10 bytes) - */ - static Bytes get_ratchet_id(const Bytes& public_bytes); - - /** - * @brief Derives a ratchet ID from an X25519PublicKey object - * @param public_key The X25519 public key object - * @return Ratchet ID (10 bytes) - */ - static Bytes get_ratchet_id(X25519PublicKey& public_key); - - public: - /** - * @brief Gets the public key bytes for announcement - * @return X25519 public key (32 bytes) - */ - Bytes public_bytes() const; - - /** - * @brief Gets the private key bytes (use with caution!) - * @return X25519 private key (32 bytes) - */ - Bytes private_bytes() const; - - /** - * @brief Gets the ratchet ID for this ratchet - * @return Ratchet ID (10 bytes) - */ - Bytes get_id() const; - - /** - * @brief Gets the creation timestamp - * @return Unix timestamp (seconds since epoch) - */ - double created_at() const { return _created_at; } - - /** - * @brief Sets the creation timestamp - * @param timestamp Unix timestamp - */ - void set_created_at(double timestamp) { _created_at = timestamp; } - - /** - * @brief Encrypts plaintext using this ratchet and peer's public key - * - * Process: - * 1. Perform X25519 ECDH: shared = ECDH(my_private, peer_public) - * 2. Derive encryption key: key = HKDF(shared, ...) - * 3. Encrypt with Fernet: ciphertext = Fernet.encrypt(key, plaintext) - * - * @param plaintext The data to encrypt - * @param peer_public_key The peer's X25519 public key (32 bytes) - * @return Encrypted ciphertext - */ - Bytes encrypt(const Bytes& plaintext, const Bytes& peer_public_key) const; - - /** - * @brief Decrypts ciphertext using this ratchet and peer's public key - * - * Process: - * 1. Perform X25519 ECDH: shared = ECDH(my_private, peer_public) - * 2. Derive decryption key: key = HKDF(shared, ...) - * 3. Decrypt with Fernet: plaintext = Fernet.decrypt(key, ciphertext) - * - * @param ciphertext The encrypted data - * @param peer_public_key The peer's X25519 public key (32 bytes) - * @return Decrypted plaintext - */ - Bytes decrypt(const Bytes& ciphertext, const Bytes& peer_public_key) const; - - /** - * @brief Derives a shared secret with peer's ratchet - * - * Performs X25519 Elliptic Curve Diffie-Hellman key exchange. - * - * @param peer_public_key The peer's X25519 public key (32 bytes) - * @return Shared secret (32 bytes) - */ - Bytes derive_shared_secret(const Bytes& peer_public_key) const; - - /** - * @brief Derives an encryption/decryption key from a shared secret - * - * Uses HKDF (HMAC-based Key Derivation Function) to derive a key suitable - * for Fernet encryption from the X25519 shared secret. - * - * @param shared_secret The result of X25519 ECDH (32 bytes) - * @return Derived encryption key (32 bytes for Fernet) - */ - Bytes derive_key(const Bytes& shared_secret) const; - - private: - Bytes _private_key; // X25519 private key (32 bytes) - Bytes _public_key; // X25519 public key (32 bytes) - double _created_at; // Unix timestamp - }; - -} } diff --git a/lib/microreticulum-shim/LXMF/LXMRouter.cpp b/lib/microreticulum-shim/LXMF/LXMRouter.cpp index 22138ca9..7d2a5ab1 100644 --- a/lib/microreticulum-shim/LXMF/LXMRouter.cpp +++ b/lib/microreticulum-shim/LXMF/LXMRouter.cpp @@ -243,19 +243,25 @@ static void static_delivery_link_established_callback(Link& link) { } } -// Static callback for resource concluded on delivery links (receiving) +// Static callback for resource concluded on delivery links (receiving). +// +// Pre-graft: the fork's Resource exposed `link()` returning the Link the +// resource was transferred over, used here to find the router that owns +// the link's destination. Vanilla upstream microReticulum @ 0.3.0 doesn't +// expose this getter — Resource is heavily refactored and Link tracking +// is internal. +// +// SPIKE STATE: this callback is a no-op. Inbound RESOURCE-form LXMF +// messages will not be dispatched to any router until either (a) +// Resource::link() is restored on upstream, or (b) LXMF is ported to +// upstream's new Resource API (which probably wires the link through +// a different callback path). Tracked in +// pyxis_microReticulum_graft_spike_findings.md. static void static_resource_concluded_callback(const Resource& resource) { - Link link = resource.link(); - if (!link) { - ERROR("static_resource_concluded_callback: Resource has no link"); - return; - } - - // Find router that owns this link's destination - RouterRegistrySlot* slot = find_router_registry_slot(link.destination().hash()); - if (slot) { - slot->router->on_resource_concluded(resource); - } + (void)resource; + ERROR("static_resource_concluded_callback: pyxis spike — Resource::link() " + "not exposed on vanilla upstream microReticulum @ 0.3.0; " + "resource-form delivery is currently disabled."); } // Static callback for outbound resource concluded (sending) @@ -1619,9 +1625,12 @@ void LXMRouter::process_sync() { if (elapsed_int != last_logged) { last_logged = elapsed_int; char buf[96]; + // Pre-graft: Link::pending_requests_count() (size_t). + // Upstream exposes pending_requests() returning std::set&, + // which we .size() here for the same value. snprintf(buf, sizeof(buf), " Sync waiting: state=%d, link_status=%d, pending_reqs=%zu, elapsed=%ds", (int)_sync_state, (int)_outbound_propagation_link.status(), - _outbound_propagation_link.pending_requests_count(), elapsed_int); + _outbound_propagation_link.pending_requests().size(), elapsed_int); INFO(buf); } } diff --git a/lib/microreticulum-shim/LXMF/LXMessage.h b/lib/microreticulum-shim/LXMF/LXMessage.h index c96e1e26..8e66ee0b 100644 --- a/lib/microreticulum-shim/LXMF/LXMessage.h +++ b/lib/microreticulum-shim/LXMF/LXMessage.h @@ -369,9 +369,12 @@ namespace LXMF { FieldEntry _fields_pool[MAX_FIELDS]; size_t _fields_count = 0; - // Destination/Source objects (may be {Type::NONE} if creating from hashes) - RNS::Destination _destination; - RNS::Destination _source; + // Destination/Source objects (may be {Type::NONE} if creating from hashes). + // Default-initialize to NONE so the LXMessage default ctor isn't + // implicitly deleted (vanilla upstream RNS::Destination has no default + // ctor — needs Type::NoneConstructor). + RNS::Destination _destination{RNS::Type::NONE}; + RNS::Destination _source{RNS::Type::NONE}; // Message metadata RNS::Bytes _hash; // Message ID (SHA256 of hashed_part) diff --git a/lib/microreticulum-shim/SegmentAccumulator.cpp b/lib/microreticulum-shim/SegmentAccumulator.cpp deleted file mode 100644 index f9566376..00000000 --- a/lib/microreticulum-shim/SegmentAccumulator.cpp +++ /dev/null @@ -1,231 +0,0 @@ -#include "SegmentAccumulator.h" -#include "Resource.h" - -using namespace RNS; -using namespace RNS::Utilities; - -SegmentAccumulator::SegmentAccumulator(AccumulatedCallback callback) - : _accumulated_callback(callback) -{ -} - -void SegmentAccumulator::set_accumulated_callback(AccumulatedCallback callback) { - _accumulated_callback = callback; -} - -void SegmentAccumulator::set_segment_callback(SegmentCallback callback) { - _segment_callback = callback; -} - -SegmentAccumulator::PendingTransferSlot* SegmentAccumulator::find_slot(const Bytes& transfer_id) { - for (size_t i = 0; i < MAX_PENDING_TRANSFERS; i++) { - if (_pending_pool[i].in_use && _pending_pool[i].transfer_id == transfer_id) { - return &_pending_pool[i]; - } - } - return nullptr; -} - -const SegmentAccumulator::PendingTransferSlot* SegmentAccumulator::find_slot(const Bytes& transfer_id) const { - for (size_t i = 0; i < MAX_PENDING_TRANSFERS; i++) { - if (_pending_pool[i].in_use && _pending_pool[i].transfer_id == transfer_id) { - return &_pending_pool[i]; - } - } - return nullptr; -} - -SegmentAccumulator::PendingTransferSlot* SegmentAccumulator::allocate_slot(const Bytes& transfer_id) { - for (size_t i = 0; i < MAX_PENDING_TRANSFERS; i++) { - if (!_pending_pool[i].in_use) { - _pending_pool[i].in_use = true; - _pending_pool[i].transfer_id = transfer_id; - return &_pending_pool[i]; - } - } - return nullptr; // Pool is full -} - -bool SegmentAccumulator::segment_completed(const Resource& resource) { - // Check if this is a multi-segment resource - if (!resource.is_segmented()) { - // Single-segment resource - caller should handle normally - return false; - } - - int segment_index = resource.segment_index(); - int total_segments = resource.total_segments(); - Bytes original_hash = resource.original_hash(); - - // Use resource hash as fallback if original_hash not set - if (!original_hash) { - original_hash = resource.hash(); - DEBUG("SegmentAccumulator: No original_hash, using resource hash as key"); - } - - std::string hash_short = original_hash.toHex().substr(0, 16); - DEBUGF("SegmentAccumulator: Received segment %d/%d for %s (%zu bytes)", - segment_index, total_segments, - hash_short.c_str(), - resource.data().size()); - - // Validate total_segments doesn't exceed our fixed array size - if (total_segments > static_cast(MAX_SEGMENTS_PER_TRANSFER)) { - ERRORF("SegmentAccumulator: Transfer has %d segments, exceeds max %zu", - total_segments, MAX_SEGMENTS_PER_TRANSFER); - return true; // We handled it (by rejecting) - } - - double now = OS::time(); - - // Find or create pending transfer - PendingTransferSlot* slot = find_slot(original_hash); - if (slot == nullptr) { - // New transfer - allocate a slot - slot = allocate_slot(original_hash); - if (slot == nullptr) { - ERRORF("SegmentAccumulator: Cannot track transfer %s, pool full (%zu max)", - hash_short.c_str(), MAX_PENDING_TRANSFERS); - return true; // We handled it (by rejecting due to pool exhaustion) - } - - // Initialize the transfer - PendingTransfer& transfer = slot->transfer; - transfer.original_hash = original_hash; - transfer.total_segments = total_segments; - transfer.received_count = 0; - transfer.segment_count = static_cast(total_segments); - transfer.started_at = now; - transfer.last_activity = now; - - // Initialize segment slots - for (int i = 0; i < total_segments; i++) { - transfer.segments[i].segment_index = i + 1; - transfer.segments[i].received = false; - } - - INFOF("SegmentAccumulator: Started tracking %d-segment transfer for %s", - total_segments, hash_short.c_str()); - } - - PendingTransfer& transfer = slot->transfer; - transfer.last_activity = now; - - // Validate segment index - if (segment_index < 1 || segment_index > transfer.total_segments) { - WARNINGF("SegmentAccumulator: Invalid segment_index %d (expected 1-%d)", - segment_index, transfer.total_segments); - return true; // We handled it (by rejecting) - } - - // Store segment data (segment_index is 1-based) - int idx = segment_index - 1; - if (!transfer.segments[idx].received) { - transfer.segments[idx].data = resource.data(); - transfer.segments[idx].data_size = resource.data().size(); - transfer.segments[idx].received = true; - transfer.received_count++; - - DEBUGF("SegmentAccumulator: Stored segment %d, %d/%d received", - segment_index, transfer.received_count, transfer.total_segments); - - // Fire per-segment callback if set - if (_segment_callback) { - _segment_callback(segment_index, total_segments, original_hash); - } - } else { - DEBUGF("SegmentAccumulator: Duplicate segment %d, ignoring", segment_index); - } - - // Check if all segments received - if (transfer.received_count == transfer.total_segments) { - std::string hash_short = original_hash.toHex().substr(0, 16); - INFOF("SegmentAccumulator: All %d segments received for %s, assembling...", - transfer.total_segments, hash_short.c_str()); - - // Assemble complete data - Bytes complete_data = assemble_segments(transfer); - - INFOF("SegmentAccumulator: Assembled %zu bytes from %d segments", - complete_data.size(), transfer.total_segments); - - // Fire accumulated callback - if (_accumulated_callback) { - _accumulated_callback(complete_data, original_hash); - } - - // Cleanup - clear the slot - slot->clear(); - } - - return true; // We handled this multi-segment resource -} - -Bytes SegmentAccumulator::assemble_segments(const PendingTransfer& transfer) { - // Calculate total size - size_t total_size = 0; - for (int i = 0; i < transfer.total_segments; i++) { - total_size += transfer.segments[i].data_size; - } - - // Concatenate in order - Bytes result; - result.reserve(total_size); - - for (int i = 0; i < transfer.total_segments; i++) { - const SegmentInfo& seg = transfer.segments[i]; - if (!seg.received) { - ERRORF("SegmentAccumulator: Missing segment %d during assembly!", i + 1); - continue; - } - result += seg.data; - } - - return result; -} - -void SegmentAccumulator::check_timeouts(double timeout_seconds) { - double now = OS::time(); - - for (size_t i = 0; i < MAX_PENDING_TRANSFERS; i++) { - if (!_pending_pool[i].in_use) { - continue; - } - - const PendingTransfer& transfer = _pending_pool[i].transfer; - double inactive_time = now - transfer.last_activity; - - if (inactive_time > timeout_seconds) { - std::string hash_short = transfer.original_hash.toHex().substr(0, 16); - WARNINGF("SegmentAccumulator: Transfer %s timed out (%.1fs inactive, %d/%d segments)", - hash_short.c_str(), - inactive_time, transfer.received_count, transfer.total_segments); - _pending_pool[i].clear(); - } - } -} - -void SegmentAccumulator::cleanup(const Bytes& original_hash) { - PendingTransferSlot* slot = find_slot(original_hash); - if (slot != nullptr) { - std::string hash_short = original_hash.toHex().substr(0, 16); - DEBUGF("SegmentAccumulator: Cleaning up transfer %s (%d/%d segments received)", - hash_short.c_str(), - slot->transfer.received_count, slot->transfer.total_segments); - slot->clear(); - } -} - -bool SegmentAccumulator::has_pending(const Bytes& original_hash) const { - return find_slot(original_hash) != nullptr; -} - -size_t SegmentAccumulator::pending_count() const { - size_t count = 0; - for (size_t i = 0; i < MAX_PENDING_TRANSFERS; i++) { - if (_pending_pool[i].in_use) { - count++; - } - } - return count; -} diff --git a/lib/microreticulum-shim/SegmentAccumulator.h b/lib/microreticulum-shim/SegmentAccumulator.h deleted file mode 100644 index e298c01d..00000000 --- a/lib/microreticulum-shim/SegmentAccumulator.h +++ /dev/null @@ -1,162 +0,0 @@ -#pragma once - -#include "Bytes.h" -#include "Log.h" -#include "Utilities/OS.h" - -#include - -namespace RNS { - - // Forward declaration to avoid circular includes - class Resource; - - /** - * SegmentAccumulator collects multi-segment resources and fires a single callback - * when all segments have been received. - * - * Python RNS splits large resources (>MAX_EFFICIENT_SIZE, ~1MB) into multiple segments. - * Each segment is transferred as a separate Resource with its own hash/proof. - * Segments share the same original_hash and have segment_index 1..total_segments. - * - * This class: - * - Tracks incoming segments by original_hash - * - Stores segment data until all segments arrive - * - Fires the accumulated callback once with the complete concatenated data - * - Handles timeout cleanup for stale transfers - */ - class SegmentAccumulator { - - public: - // Fixed-size pool limits to eliminate heap fragmentation - static constexpr size_t MAX_PENDING_TRANSFERS = 8; - static constexpr size_t MAX_SEGMENTS_PER_TRANSFER = 64; - - // Callback fired when all segments are received - // Parameters: complete_data, original_hash - using AccumulatedCallback = std::function; - - // Callback for individual segment completion (optional, for progress tracking) - using SegmentCallback = std::function; - - public: - SegmentAccumulator() = default; - explicit SegmentAccumulator(AccumulatedCallback callback); - - /** - * Set the callback for completed multi-segment resources. - */ - void set_accumulated_callback(AccumulatedCallback callback); - - /** - * Set optional per-segment progress callback. - */ - void set_segment_callback(SegmentCallback callback); - - /** - * Called when a Resource segment completes. - * - * @param resource The completed resource (may be single or multi-segment) - * @return true if this was a multi-segment resource that was handled, - * false if it was a single-segment resource (caller should invoke normal callback) - */ - bool segment_completed(const Resource& resource); - - /** - * Check for timed-out transfers and clean them up. - * Should be called periodically (e.g., from watchdog). - * - * @param timeout_seconds Maximum time since last activity before cleanup - */ - void check_timeouts(double timeout_seconds = 600.0); - - /** - * Manually cleanup a specific transfer. - */ - void cleanup(const Bytes& original_hash); - - /** - * Check if a transfer is in progress for the given original_hash. - */ - bool has_pending(const Bytes& original_hash) const; - - /** - * Get the number of pending (incomplete) transfers. - */ - size_t pending_count() const; - - private: - struct SegmentInfo { - int segment_index = 0; - size_t data_size = 0; - Bytes data; - bool received = false; - - void clear() { - segment_index = 0; - data_size = 0; - data.clear(); - received = false; - } - }; - - struct PendingTransfer { - Bytes original_hash; - int total_segments = 0; - int received_count = 0; - SegmentInfo segments[MAX_SEGMENTS_PER_TRANSFER]; // Fixed array instead of std::vector - size_t segment_count = 0; - double started_at = 0.0; - double last_activity = 0.0; - - void clear() { - original_hash.clear(); - total_segments = 0; - received_count = 0; - for (size_t i = 0; i < segment_count; i++) { - segments[i].clear(); - } - segment_count = 0; - started_at = 0.0; - last_activity = 0.0; - } - }; - - struct PendingTransferSlot { - bool in_use = false; - Bytes transfer_id; // key (original_hash) - PendingTransfer transfer; - - void clear() { - in_use = false; - transfer_id.clear(); - transfer.clear(); - } - }; - - // Fixed-size pool instead of std::map - PendingTransferSlot _pending_pool[MAX_PENDING_TRANSFERS]; - - AccumulatedCallback _accumulated_callback = nullptr; - SegmentCallback _segment_callback = nullptr; - - /** - * Find a slot by transfer_id (original_hash). - * @return pointer to slot if found, nullptr otherwise - */ - PendingTransferSlot* find_slot(const Bytes& transfer_id); - const PendingTransferSlot* find_slot(const Bytes& transfer_id) const; - - /** - * Allocate a new slot for a transfer. - * @return pointer to slot if available, nullptr if pool is full - */ - PendingTransferSlot* allocate_slot(const Bytes& transfer_id); - - /** - * Concatenate all segments in order and return complete data. - */ - Bytes assemble_segments(const PendingTransfer& transfer); - }; - -}