dandri/pyxis
1
0
Fork 0
mirror of https://github.com/torlando-tech/pyxis.git synced 2026-08-26 20:50:00 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Track A: pyxis builds against attermann/microReticulum @ 0.3.0 + fixes

Browse Source 
🎯 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 `<X25519.h>`
                                     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.
This commit is contained in:
torlando-tech
2026-05-05 01:44:50 -04:00
parent 02ceeda75b
commit fd2d3de359
9 changed files with 28 additions and 1719 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/microreticulum-shim/Buffer.cpp
-404
View File
@@ -1,404 +0,0 @@
#include "Buffer.h"
#include "ChannelData.h" // Required for Channel template instantiation
#include "Cryptography/BZ2.h"
#include <algorithm>
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<StreamDataMessage&>(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<RawChannelReader::ReadyCallback> _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<uint8_t>((header >> 8) & 0xFF));
result.append(static_cast<uint8_t>(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<uint16_t>(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<RawChannelReaderData>(stream_id, channel)) {
// Register StreamDataMessage type (as system message)
_object->_channel->register_message_type<StreamDataMessage>(true);
// Add message handler - capture shared_ptr to keep data alive
// and use weak_ptr to avoid preventing destruction
std::weak_ptr<RawChannelReaderData> 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<StreamDataMessage>(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<RawChannelReader, RawChannelWriter>
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
lib/microreticulum-shim/Buffer.h
-163
View File
@@ -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 <memory>
#include <functional>
#include <vector>
#include <utility>
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<void(size_t)>;
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<RawChannelReaderData> _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<RawChannelReader, RawChannelWriter>
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
lib/microreticulum-shim/ChannelData.h
-369
View File
@@ -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 <map>
#include <vector>
#include <functional>
#include <memory>
#include <utility> // 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<MessageBase>& message() { return _message; }
void set_message(std::unique_ptr<MessageBase> 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<uint16_t>(data[0]) << 8) | data[1];
// SEQUENCE (2 bytes, big-endian)
uint16_t sequence = (static_cast<uint16_t>(data[2]) << 8) | data[3];
// LENGTH (2 bytes, big-endian)
uint16_t length = (static_cast<uint16_t>(data[4]) << 8) | data[5];
// Validate length
if (wire_data.size() < 6 + static_cast<size_t>(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<MessageBase> _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<int32_t>(new_seq) - static_cast<int32_t>(existing_seq);
if (diff >= static_cast<int32_t>(Type::Channel::SEQ_MODULUS / 2)) {
diff -= Type::Channel::SEQ_MODULUS;
} else if (diff < -static_cast<int32_t>(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<typename Func>
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<typename Func>
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<uint16_t, std::function<std::unique_ptr<MessageBase>()>> _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<std::function<bool(MessageBase&)>> _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
lib/microreticulum-shim/Cryptography/Ratchet.cpp
-193
View File
@@ -1,193 +0,0 @@
#include "Ratchet.h"
#include "Fernet.h"
#include <Utilities/OS.h>
#include <Curve25519.h>
#include <SHA256.h>
#include <stdexcept>
#include <cstring>
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;
}
}
lib/microreticulum-shim/Cryptography/Ratchet.h
-181
View File
@@ -1,181 +0,0 @@
#pragma once
#include <Bytes.h>
#include <Type.h>
#include <Log.h>
#include "X25519.h"
#include "HKDF.h"
#include "Fernet.h"
#include "Hashes.h"
#include <memory>
#include <stdint.h>
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<Ratchet>;
// 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
};
} }
lib/microreticulum-shim/LXMF/LXMRouter.cpp
+22 -13
View File
@@ -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) { static void static_resource_concluded_callback(const Resource& resource) {
Link link = resource.link(); (void)resource;
if (!link) { ERROR("static_resource_concluded_callback: pyxis spike — Resource::link() "
ERROR("static_resource_concluded_callback: Resource has no link"); "not exposed on vanilla upstream microReticulum @ 0.3.0; "
return; "resource-form delivery is currently disabled.");
}
// 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);
}
} }
// Static callback for outbound resource concluded (sending) // Static callback for outbound resource concluded (sending)
@@ -1619,9 +1625,12 @@ void LXMRouter::process_sync() {
if (elapsed_int != last_logged) { if (elapsed_int != last_logged) {
last_logged = elapsed_int; last_logged = elapsed_int;
char buf[96]; 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", snprintf(buf, sizeof(buf), " Sync waiting: state=%d, link_status=%d, pending_reqs=%zu, elapsed=%ds",
(int)_sync_state, (int)_outbound_propagation_link.status(), (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); INFO(buf);
} }
} }
lib/microreticulum-shim/LXMF/LXMessage.h
+6 -3
View File
@@ -369,9 +369,12 @@ namespace LXMF {
FieldEntry _fields_pool[MAX_FIELDS]; FieldEntry _fields_pool[MAX_FIELDS];
size_t _fields_count = 0; size_t _fields_count = 0;
// Destination/Source objects (may be {Type::NONE} if creating from hashes) // Destination/Source objects (may be {Type::NONE} if creating from hashes).
RNS::Destination _destination; // Default-initialize to NONE so the LXMessage default ctor isn't
RNS::Destination _source; // 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 // Message metadata
RNS::Bytes _hash; // Message ID (SHA256 of hashed_part) RNS::Bytes _hash; // Message ID (SHA256 of hashed_part)
lib/microreticulum-shim/SegmentAccumulator.cpp
-231
View File
@@ -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<int>(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<size_t>(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;
}
lib/microreticulum-shim/SegmentAccumulator.h
-162
View File
@@ -1,162 +0,0 @@
#pragma once
#include "Bytes.h"
#include "Log.h"
#include "Utilities/OS.h"
#include <functional>
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<void(const Bytes& data, const Bytes& original_hash)>;
// Callback for individual segment completion (optional, for progress tracking)
using SegmentCallback = std::function<void(int segment_index, int total_segments, const Bytes& original_hash)>;
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);
};
}