feat: initial meshservice implementation
Generic decentralized service layer for mesh networks. Includes FAPP (psychotherapy) and Housing as reference services. - ServiceMessage with Ed25519 signatures - Wire protocol (64-byte header + CBOR) - ServiceRouter with pluggable handlers - Verification framework (3 trust levels) - 32 tests, 3 examples
This commit is contained in:
406
src/store.rs
Normal file
406
src/store.rs
Normal file
@@ -0,0 +1,406 @@
|
||||
//! In-memory message store with eviction policies.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
use crate::message::ServiceMessage;
|
||||
|
||||
/// Configuration for the message store.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct StoreConfig {
|
||||
/// Maximum messages per service.
|
||||
pub max_per_service: usize,
|
||||
/// Maximum messages per sender (per service).
|
||||
pub max_per_sender: usize,
|
||||
/// Maximum total messages.
|
||||
pub max_total: usize,
|
||||
/// Prune interval in seconds.
|
||||
pub prune_interval_secs: u64,
|
||||
}
|
||||
|
||||
impl Default for StoreConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
max_per_service: 10_000,
|
||||
max_per_sender: 100,
|
||||
max_total: 50_000,
|
||||
prune_interval_secs: 300,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A stored message with metadata.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct StoredMessage {
|
||||
pub message: ServiceMessage,
|
||||
/// Sender's public key (needed for verification).
|
||||
pub sender_public_key: [u8; 32],
|
||||
/// When we stored this message.
|
||||
pub stored_at: u64,
|
||||
}
|
||||
|
||||
/// Generic service message store.
|
||||
///
|
||||
/// Organized by service_id, then by sender_address, then by message_id.
|
||||
pub struct ServiceStore {
|
||||
config: StoreConfig,
|
||||
/// service_id -> sender_address -> message_id -> StoredMessage
|
||||
messages: HashMap<u32, HashMap<[u8; 16], HashMap<[u8; 16], StoredMessage>>>,
|
||||
/// Total message count.
|
||||
total_count: usize,
|
||||
/// Last prune timestamp.
|
||||
last_prune: u64,
|
||||
}
|
||||
|
||||
impl ServiceStore {
|
||||
/// Create a new store with default config.
|
||||
pub fn new() -> Self {
|
||||
Self::with_config(StoreConfig::default())
|
||||
}
|
||||
|
||||
/// Create with custom config.
|
||||
pub fn with_config(config: StoreConfig) -> Self {
|
||||
Self {
|
||||
config,
|
||||
messages: HashMap::new(),
|
||||
total_count: 0,
|
||||
last_prune: 0,
|
||||
}
|
||||
}
|
||||
|
||||
/// Store a message, returning true if it was new.
|
||||
pub fn store(&mut self, message: ServiceMessage, sender_public_key: [u8; 32]) -> bool {
|
||||
// Prune if interval passed
|
||||
self.maybe_prune();
|
||||
|
||||
let service_id = message.service_id;
|
||||
let sender_address = message.sender_address;
|
||||
let message_id = message.id;
|
||||
|
||||
// Check per-service limit and evict if needed
|
||||
{
|
||||
let service_count: usize = self.messages
|
||||
.get(&service_id)
|
||||
.map(|s| s.values().map(|m| m.len()).sum())
|
||||
.unwrap_or(0);
|
||||
if service_count >= self.config.max_per_service {
|
||||
self.evict_oldest_in_service(service_id);
|
||||
}
|
||||
}
|
||||
|
||||
// Check per-sender limit and evict if needed
|
||||
{
|
||||
let sender_count = self.messages
|
||||
.get(&service_id)
|
||||
.and_then(|s| s.get(&sender_address))
|
||||
.map(|m| m.len())
|
||||
.unwrap_or(0);
|
||||
if sender_count >= self.config.max_per_sender {
|
||||
self.evict_oldest_from_sender(service_id, sender_address);
|
||||
}
|
||||
}
|
||||
|
||||
// Get or create maps
|
||||
let service_map = self.messages.entry(service_id).or_default();
|
||||
let sender_map = service_map.entry(sender_address).or_default();
|
||||
|
||||
// Check for existing message
|
||||
let is_new_or_update = if let Some(existing) = sender_map.get(&message_id) {
|
||||
// Existing: only update if higher sequence
|
||||
if message.sequence <= existing.message.sequence {
|
||||
return false;
|
||||
}
|
||||
// This is an update, not a new message
|
||||
false
|
||||
} else {
|
||||
// New message
|
||||
true
|
||||
};
|
||||
|
||||
let stored_at = now();
|
||||
sender_map.insert(
|
||||
message_id,
|
||||
StoredMessage {
|
||||
message,
|
||||
sender_public_key,
|
||||
stored_at,
|
||||
},
|
||||
);
|
||||
|
||||
if is_new_or_update {
|
||||
self.total_count += 1;
|
||||
}
|
||||
|
||||
// Return true for both new messages and updates
|
||||
true
|
||||
}
|
||||
|
||||
/// Get a message by service, sender, and ID.
|
||||
pub fn get(
|
||||
&self,
|
||||
service_id: u32,
|
||||
sender_address: &[u8; 16],
|
||||
message_id: &[u8; 16],
|
||||
) -> Option<&StoredMessage> {
|
||||
self.messages
|
||||
.get(&service_id)?
|
||||
.get(sender_address)?
|
||||
.get(message_id)
|
||||
}
|
||||
|
||||
/// Get all messages from a sender in a service.
|
||||
pub fn by_sender(&self, service_id: u32, sender_address: &[u8; 16]) -> Vec<&StoredMessage> {
|
||||
self.messages
|
||||
.get(&service_id)
|
||||
.and_then(|s| s.get(sender_address))
|
||||
.map(|m| m.values().collect())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
/// Get all messages in a service.
|
||||
pub fn by_service(&self, service_id: u32) -> Vec<&StoredMessage> {
|
||||
self.messages
|
||||
.get(&service_id)
|
||||
.map(|s| s.values().flat_map(|m| m.values()).collect())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
/// Query messages with a predicate.
|
||||
pub fn query<F>(&self, service_id: u32, predicate: F) -> Vec<&StoredMessage>
|
||||
where
|
||||
F: Fn(&StoredMessage) -> bool,
|
||||
{
|
||||
self.by_service(service_id)
|
||||
.into_iter()
|
||||
.filter(|m| predicate(m))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Remove a specific message.
|
||||
pub fn remove(
|
||||
&mut self,
|
||||
service_id: u32,
|
||||
sender_address: &[u8; 16],
|
||||
message_id: &[u8; 16],
|
||||
) -> Option<StoredMessage> {
|
||||
let result = self
|
||||
.messages
|
||||
.get_mut(&service_id)?
|
||||
.get_mut(sender_address)?
|
||||
.remove(message_id);
|
||||
|
||||
if result.is_some() {
|
||||
self.total_count = self.total_count.saturating_sub(1);
|
||||
}
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
/// Remove all messages from a sender.
|
||||
pub fn remove_sender(&mut self, service_id: u32, sender_address: &[u8; 16]) -> usize {
|
||||
let count = self
|
||||
.messages
|
||||
.get_mut(&service_id)
|
||||
.and_then(|s| s.remove(sender_address))
|
||||
.map(|m| m.len())
|
||||
.unwrap_or(0);
|
||||
|
||||
self.total_count = self.total_count.saturating_sub(count);
|
||||
count
|
||||
}
|
||||
|
||||
/// Prune expired messages.
|
||||
pub fn prune_expired(&mut self) -> usize {
|
||||
let now = now();
|
||||
let mut removed = 0;
|
||||
|
||||
for service_map in self.messages.values_mut() {
|
||||
for sender_map in service_map.values_mut() {
|
||||
let expired: Vec<[u8; 16]> = sender_map
|
||||
.iter()
|
||||
.filter(|(_, m)| m.message.is_expired())
|
||||
.map(|(id, _)| *id)
|
||||
.collect();
|
||||
|
||||
for id in expired {
|
||||
sender_map.remove(&id);
|
||||
removed += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
self.total_count = self.total_count.saturating_sub(removed);
|
||||
self.last_prune = now;
|
||||
removed
|
||||
}
|
||||
|
||||
/// Get total message count.
|
||||
pub fn len(&self) -> usize {
|
||||
self.total_count
|
||||
}
|
||||
|
||||
/// Check if empty.
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.total_count == 0
|
||||
}
|
||||
|
||||
/// Get count by service.
|
||||
pub fn service_count(&self, service_id: u32) -> usize {
|
||||
self.messages
|
||||
.get(&service_id)
|
||||
.map(|s| s.values().map(|m| m.len()).sum())
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
/// Run prune if interval passed.
|
||||
fn maybe_prune(&mut self) {
|
||||
let now = now();
|
||||
if now.saturating_sub(self.last_prune) >= self.config.prune_interval_secs {
|
||||
self.prune_expired();
|
||||
}
|
||||
}
|
||||
|
||||
/// Evict oldest message in a service.
|
||||
fn evict_oldest_in_service(&mut self, service_id: u32) {
|
||||
let Some(service_map) = self.messages.get_mut(&service_id) else {
|
||||
return;
|
||||
};
|
||||
|
||||
let mut oldest: Option<([u8; 16], [u8; 16], u64)> = None;
|
||||
|
||||
for (sender, msgs) in service_map.iter() {
|
||||
for (id, stored) in msgs.iter() {
|
||||
match oldest {
|
||||
Some((_, _, ts)) if stored.message.timestamp < ts => {
|
||||
oldest = Some((*sender, *id, stored.message.timestamp));
|
||||
}
|
||||
None => {
|
||||
oldest = Some((*sender, *id, stored.message.timestamp));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if let Some((sender, id, _)) = oldest {
|
||||
if let Some(sender_map) = service_map.get_mut(&sender) {
|
||||
sender_map.remove(&id);
|
||||
self.total_count = self.total_count.saturating_sub(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Evict oldest message from a sender.
|
||||
fn evict_oldest_from_sender(&mut self, service_id: u32, sender_address: [u8; 16]) {
|
||||
let Some(sender_map) = self
|
||||
.messages
|
||||
.get_mut(&service_id)
|
||||
.and_then(|s| s.get_mut(&sender_address))
|
||||
else {
|
||||
return;
|
||||
};
|
||||
|
||||
let oldest = sender_map
|
||||
.iter()
|
||||
.min_by_key(|(_, m)| m.message.timestamp)
|
||||
.map(|(id, _)| *id);
|
||||
|
||||
if let Some(id) = oldest {
|
||||
sender_map.remove(&id);
|
||||
self.total_count = self.total_count.saturating_sub(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for ServiceStore {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
fn now() -> u64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_secs()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::{identity::ServiceIdentity, message::ServiceMessage, service_ids::FAPP};
|
||||
|
||||
fn make_message(id: &ServiceIdentity, seq: u64) -> ServiceMessage {
|
||||
ServiceMessage::announce(id, FAPP, b"test".to_vec(), seq)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn store_and_retrieve() {
|
||||
let mut store = ServiceStore::new();
|
||||
let id = ServiceIdentity::generate();
|
||||
let msg = make_message(&id, 1);
|
||||
|
||||
assert!(store.store(msg.clone(), id.public_key()));
|
||||
assert_eq!(store.len(), 1);
|
||||
|
||||
let retrieved = store.get(FAPP, &id.address(), &msg.id);
|
||||
assert!(retrieved.is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn duplicate_rejected() {
|
||||
let mut store = ServiceStore::new();
|
||||
let id = ServiceIdentity::generate();
|
||||
let msg = make_message(&id, 1);
|
||||
|
||||
assert!(store.store(msg.clone(), id.public_key()));
|
||||
assert!(!store.store(msg.clone(), id.public_key())); // Duplicate
|
||||
assert_eq!(store.len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn higher_sequence_updates() {
|
||||
let mut store = ServiceStore::new();
|
||||
let id = ServiceIdentity::generate();
|
||||
let msg1 = make_message(&id, 1);
|
||||
let mut msg2 = make_message(&id, 2);
|
||||
msg2.id = msg1.id; // Same ID
|
||||
|
||||
store.store(msg1.clone(), id.public_key());
|
||||
assert!(store.store(msg2.clone(), id.public_key())); // Updates
|
||||
|
||||
let retrieved = store.get(FAPP, &id.address(), &msg1.id).unwrap();
|
||||
assert_eq!(retrieved.message.sequence, 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn query_by_sender() {
|
||||
let mut store = ServiceStore::new();
|
||||
let id1 = ServiceIdentity::generate();
|
||||
let id2 = ServiceIdentity::generate();
|
||||
|
||||
store.store(make_message(&id1, 1), id1.public_key());
|
||||
store.store(make_message(&id1, 2), id1.public_key());
|
||||
store.store(make_message(&id2, 1), id2.public_key());
|
||||
|
||||
let sender1_msgs = store.by_sender(FAPP, &id1.address());
|
||||
assert_eq!(sender1_msgs.len(), 2);
|
||||
|
||||
let sender2_msgs = store.by_sender(FAPP, &id2.address());
|
||||
assert_eq!(sender2_msgs.len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn remove_sender() {
|
||||
let mut store = ServiceStore::new();
|
||||
let id = ServiceIdentity::generate();
|
||||
|
||||
store.store(make_message(&id, 1), id.public_key());
|
||||
store.store(make_message(&id, 2), id.public_key());
|
||||
assert_eq!(store.len(), 2);
|
||||
|
||||
let removed = store.remove_sender(FAPP, &id.address());
|
||||
assert_eq!(removed, 2);
|
||||
assert_eq!(store.len(), 0);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user