feat: cluster test improvements; datastore pools (#45)

Introduce a gossip-converged pooled datastore protocol layered on SWIM piggybacking, backed by a reusable gossip-channel abstraction, plus a Docker-free multi-process cluster test runner.

- crates/distribution/src/gossip_channel.rs: add the GossipChannel trait (piggyback on SWIM messages) and a budget-limited DisseminationBuffer<T> that replaces the four duplicated Lambda*ceil(log2(n)) dissemination copies
- crates/datastore/src/pool: add PoolDisseminator (CRDT state for membership/capacity/content-location/ACL with join/leave/announce) and PoolCoordinator (placement-aware CRUD actor delegating to the co-located DatastoreNode)
- crates/shared-types/src/pool.rs: add shared pool protocol types (PoolId plus member/capacity/content-location/ACL entries and PoolConfig) consumed by both distribution and datastore
- crates/dashboard/src/pool_html.rs: add a live pool dashboard page (membership, capacity, content locations) and add pool_tests integration coverage
- xtask/src/sim_cluster.rs: add the sim-cluster runner that spawns N swactor nodes over iroh through a local relay server, reusing the docker cluster scenarios without Docker
- crates/distribution/src/iroh_driver.rs: add relay-URL resolution (cache, then SWIM gossip, then home relay) with a 2s connect timeout to back the relay-based connections

Signed-off-by: Zachery Aaron Shores-Chmielewski <zacheryasc@gmail.com>
This commit is contained in:
Zachery Aaron Shores-Chmielewski 2026-02-20 17:30:37 +00:00 committed by zacheryasc
parent c6cb88e335
commit 599678e7c9
18 changed files with 3586 additions and 6 deletions

2
.gitignore vendored
View file

@ -25,3 +25,5 @@ deploy.toml
# Local dev node state # Local dev node state
.dev-node/ .dev-node/
.dev-cluster/
.sim-cluster/

3
Cargo.lock generated
View file

@ -6949,9 +6949,12 @@ name = "xtask"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"clap", "clap",
"iroh-relay",
"libc", "libc",
"reqwest 0.12.28",
"serde", "serde",
"serde_json", "serde_json",
"tokio",
"toml 0.8.23", "toml 0.8.23",
] ]

View file

@ -0,0 +1,392 @@
pub const POOL_HTML: &str = r##"<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Swactor Runtime – Pool</title>
<style>
* { margin: 0; padding: 0; box-sizing: border-box; }
body { font-family: 'Menlo', 'Consolas', 'Monaco', monospace; background: #0f1117; color: #e0e0e0; font-size: 13px; }
.header {
display: flex; align-items: center; justify-content: space-between;
padding: 12px 20px; background: #161822; border-bottom: 1px solid #2a2d3e;
}
.header-left { display: flex; align-items: center; }
.header h1 { font-size: 16px; font-weight: 600; color: #fff; }
.status-dot {
width: 10px; height: 10px; border-radius: 50%; background: #4caf50;
display: inline-block; margin-left: 8px; vertical-align: middle;
}
.status-dot.disconnected { background: #f44336; }
.status-dot.done { background: #ff9800; }
.nav-links { display: flex; gap: 4px; margin-left: 20px; }
.nav-link {
color: #888; text-decoration: none; font-size: 12px;
padding: 4px 10px; border-radius: 3px; transition: color 0.2s;
}
.nav-link:hover { color: #e0e0e0; }
.nav-link.active { color: #fff; background: #2a2d3e; }
.header-right { display: flex; align-items: center; gap: 12px; }
.grid {
display: grid;
grid-template-columns: 1fr 1fr;
gap: 12px; padding: 12px;
}
.panel {
background: #161822; border: 1px solid #2a2d3e; border-radius: 6px;
padding: 14px; overflow: hidden;
}
.panel h2 { font-size: 12px; color: #888; text-transform: uppercase; letter-spacing: 1px; margin-bottom: 10px; }
.full-width { grid-column: 1 / -1; }
.stats-cards {
display: grid; grid-template-columns: repeat(5, 1fr); gap: 10px;
}
.stat-card {
background: #1c1f2e; border-radius: 4px; padding: 10px; text-align: center;
}
.stat-card .value { font-size: 22px; font-weight: 700; color: #fff; }
.stat-card .label { font-size: 10px; color: #888; text-transform: uppercase; margin-top: 2px; }
.members-table { max-height: 400px; overflow-y: auto; }
.members-table table { width: 100%; border-collapse: collapse; }
.members-table th, .members-table td {
padding: 4px 8px; text-align: left; border-bottom: 1px solid #1c1f2e; font-size: 11px;
white-space: nowrap;
}
.members-table th { color: #888; font-weight: 500; position: sticky; top: 0; background: #161822; }
.content-table { max-height: 400px; overflow-y: auto; }
.content-table table { width: 100%; border-collapse: collapse; }
.content-table th, .content-table td {
padding: 4px 8px; text-align: left; border-bottom: 1px solid #1c1f2e; font-size: 11px;
white-space: nowrap;
}
.content-table th { color: #888; font-weight: 500; position: sticky; top: 0; background: #161822; }
/* Capacity bars */
.cap-bar-wrap { margin-bottom: 6px; }
.cap-bar-label {
display: flex; justify-content: space-between; font-size: 10px; color: #888; margin-bottom: 2px;
}
.cap-bar {
height: 14px; background: #1c1f2e; border-radius: 3px; overflow: hidden;
}
.cap-bar-fill {
height: 100%; background: #6366f1; border-radius: 3px;
transition: width 0.3s ease;
}
.cap-bar-fill.warn { background: #ff9800; }
.cap-bar-fill.crit { background: #f44336; }
/* Buttons */
button {
background: #1e2030; color: #e0e0e0; border: 1px solid #2a2d3e;
border-radius: 4px; padding: 6px 14px; font-family: inherit;
font-size: 13px; cursor: pointer; min-height: 38px;
transition: border-color 0.15s;
}
button:hover { border-color: #6366f1; color: #fff; }
button:disabled { opacity: 0.4; cursor: default; }
button.primary { background: #6366f1; border-color: #6366f1; color: #fff; font-weight: 600; }
button.primary:hover { background: #5558e6; }
button.danger:hover { border-color: #f44336; }
/* No-pool message */
.no-pool {
display: flex; align-items: center; justify-content: center;
height: calc(100vh - 49px); color: #555; font-size: 16px;
flex-direction: column; gap: 8px;
}
/* Toast */
.toast {
position: fixed; bottom: 20px; right: 20px; padding: 10px 16px;
border-radius: 4px; font-size: 12px; z-index: 100; opacity: 0;
transition: opacity 0.3s; pointer-events: none;
}
.toast.show { opacity: 1; }
.toast.success { background: #4caf50; color: #fff; }
.toast.error { background: #f44336; color: #fff; }
::-webkit-scrollbar { width: 6px; }
::-webkit-scrollbar-track { background: #0f1117; }
::-webkit-scrollbar-thumb { background: #2a2d3e; border-radius: 3px; }
</style>
</head>
<body>
<div class="header">
<div class="header-left">
<h1>
Swactor Runtime Dashboard
<span id="statusDot" class="status-dot disconnected"></span>
</h1>
<nav class="nav-links">
<a href="/" class="nav-link">Overview</a>
<a href="/actors" class="nav-link">Actors</a>
<a href="/distribution" class="nav-link">Distribution</a>
<a href="/datastore" class="nav-link">Datastore</a>
<a href="/pool" class="nav-link active">Pool</a>
</nav>
</div>
<div class="header-right">
<button id="joinBtn" class="primary" onclick="joinPool()" style="display:none">Join Pool</button>
<button id="leaveBtn" class="danger" onclick="leavePool()" style="display:none">Leave Pool</button>
</div>
</div>
<div id="noPool" class="no-pool" style="display:none">
<div>No pool configured</div>
<div style="font-size:12px;color:#444;">Start the node with --pool-name to enable pooled storage</div>
</div>
<div id="poolContent" style="display:none">
<div class="grid">
<!-- Summary cards -->
<div class="panel full-width">
<h2>Pool Summary</h2>
<div class="stats-cards">
<div class="stat-card"><div class="value" id="statPoolName">-</div><div class="label">Pool Name</div></div>
<div class="stat-card"><div class="value" id="statMembers">0</div><div class="label">Members</div></div>
<div class="stat-card"><div class="value" id="statContent">0</div><div class="label">Content Items</div></div>
<div class="stat-card"><div class="value" id="statTotal">0</div><div class="label">Total Capacity</div></div>
<div class="stat-card"><div class="value" id="statUsed">0</div><div class="label">Used</div></div>
</div>
</div>
<!-- Capacity chart -->
<div class="panel">
<h2>Capacity by Node</h2>
<div id="capacityChart"></div>
<div id="capEmpty" style="color:#555;font-size:11px;">No members yet</div>
</div>
<!-- Members table -->
<div class="panel">
<h2>Members <span id="memberCount" style="color:#555;font-weight:400;"></span></h2>
<div class="members-table">
<table>
<thead><tr><th>Node ID</th><th>State</th><th>Total</th><th>Used</th><th>Free</th></tr></thead>
<tbody id="membersBody"></tbody>
</table>
</div>
</div>
<!-- Content locations -->
<div class="panel full-width">
<h2>Content Location Map <span id="contentCount" style="color:#555;font-weight:400;"></span></h2>
<div class="content-table">
<table>
<thead><tr><th>Content Hash</th><th>Replicas</th><th>Nodes</th></tr></thead>
<tbody id="contentBody"></tbody>
</table>
</div>
</div>
<!-- ACL panel -->
<div class="panel full-width">
<h2>Access Control <span id="aclMode" style="color:#555;font-weight:400;"></span></h2>
<div id="aclOpen" style="color:#555;font-size:11px;">Open mode &mdash; any node may join the pool</div>
<div id="aclTable" class="content-table" style="display:none">
<table>
<thead><tr><th>Node ID</th><th>Granted By</th><th>Status</th></tr></thead>
<tbody id="aclBody"></tbody>
</table>
</div>
</div>
</div>
</div>
<div class="toast" id="toast"></div>
<script>
(function() {
var dot = document.getElementById('statusDot');
var poolConfigured = false;
function $(id) { return document.getElementById(id); }
function formatBytes(b) {
if (b === 0) return '0 B';
var units = ['B', 'KB', 'MB', 'GB', 'TB'];
var i = Math.floor(Math.log(b) / Math.log(1024));
if (i >= units.length) i = units.length - 1;
return (b / Math.pow(1024, i)).toFixed(i > 0 ? 1 : 0) + ' ' + units[i];
}
function escapeHtml(s) {
if (!s) return '';
return s.replace(/&/g,'&amp;').replace(/</g,'&lt;').replace(/>/g,'&gt;');
}
window.toast = function(msg, type) {
var t = $('toast');
t.textContent = msg;
t.className = 'toast show ' + type;
setTimeout(function() { t.className = 'toast'; }, 2500);
};
function updatePool(snap) {
if (!snap) {
if (!poolConfigured) {
$('noPool').style.display = 'flex';
$('poolContent').style.display = 'none';
$('joinBtn').style.display = 'none';
$('leaveBtn').style.display = 'none';
}
return;
}
poolConfigured = true;
$('noPool').style.display = 'none';
$('poolContent').style.display = 'block';
$('joinBtn').style.display = 'inline-block';
$('leaveBtn').style.display = 'inline-block';
// Summary cards
$('statPoolName').textContent = snap.pool_name || '-';
$('statPoolName').style.fontSize = '14px';
$('statMembers').textContent = snap.member_count;
$('statContent').textContent = snap.content_count;
$('statTotal').textContent = formatBytes(snap.total_bytes);
$('statUsed').textContent = formatBytes(snap.used_bytes);
// Members table
var mbody = $('membersBody');
mbody.innerHTML = '';
$('memberCount').textContent = '(' + snap.members.length + ')';
for (var i = 0; i < snap.members.length; i++) {
var m = snap.members[i];
var free = m.total_bytes - m.used_bytes;
var tr = document.createElement('tr');
tr.innerHTML =
'<td style="color:#6366f1;font-size:11px;" title="' + escapeHtml(m.node_id) + '">' + m.node_id.substring(0, 16) + '\u2026</td>' +
'<td style="color:#4caf50;">' + escapeHtml(m.state) + '</td>' +
'<td style="color:#888;">' + formatBytes(m.total_bytes) + '</td>' +
'<td style="color:#888;">' + formatBytes(m.used_bytes) + '</td>' +
'<td style="color:#888;">' + formatBytes(free) + '</td>';
mbody.appendChild(tr);
}
// Capacity chart
var chart = $('capacityChart');
chart.innerHTML = '';
var capEmpty = $('capEmpty');
if (snap.members.length === 0) {
capEmpty.style.display = 'block';
} else {
capEmpty.style.display = 'none';
for (var i = 0; i < snap.members.length; i++) {
var m = snap.members[i];
var pct = m.total_bytes > 0 ? Math.round((m.used_bytes / m.total_bytes) * 100) : 0;
var fillClass = 'cap-bar-fill';
if (pct > 90) fillClass += ' crit';
else if (pct > 70) fillClass += ' warn';
var wrap = document.createElement('div');
wrap.className = 'cap-bar-wrap';
wrap.innerHTML =
'<div class="cap-bar-label"><span>' + m.node_id.substring(0, 12) + '\u2026</span><span>' + pct + '% (' + formatBytes(m.used_bytes) + ' / ' + formatBytes(m.total_bytes) + ')</span></div>' +
'<div class="cap-bar"><div class="' + fillClass + '" style="width:' + pct + '%"></div></div>';
chart.appendChild(wrap);
}
}
// Content locations
var cbody = $('contentBody');
cbody.innerHTML = '';
$('contentCount').textContent = '(' + snap.content_locations.length + ')';
for (var i = 0; i < snap.content_locations.length; i++) {
var cl = snap.content_locations[i];
var nodeList = cl.nodes.map(function(n) { return n.substring(0, 12) + '\u2026'; }).join(', ');
var tr = document.createElement('tr');
tr.innerHTML =
'<td style="color:#6366f1;font-size:11px;" title="' + escapeHtml(cl.content_hash) + '">' + cl.content_hash.substring(0, 16) + '\u2026</td>' +
'<td>' + cl.replica_count + '</td>' +
'<td style="color:#888;font-size:10px;">' + escapeHtml(nodeList) + '</td>';
cbody.appendChild(tr);
}
// ACL panel
var aclMode = snap.acl_mode || 'open';
var aclEntries = snap.acl || [];
$('aclMode').textContent = '(' + aclMode + ')';
if (aclMode === 'open' || aclEntries.length === 0) {
$('aclOpen').style.display = 'block';
$('aclTable').style.display = 'none';
} else {
$('aclOpen').style.display = 'none';
$('aclTable').style.display = 'block';
var abody = $('aclBody');
abody.innerHTML = '';
for (var i = 0; i < aclEntries.length; i++) {
var a = aclEntries[i];
var status = a.revoked ? 'revoked' : 'granted';
var statusColor = a.revoked ? '#f44336' : '#4caf50';
var tr = document.createElement('tr');
tr.innerHTML =
'<td style="color:#6366f1;font-size:11px;" title="' + escapeHtml(a.node_id) + '">' + a.node_id.substring(0, 16) + '\u2026</td>' +
'<td style="color:#888;font-size:10px;" title="' + escapeHtml(a.granted_by) + '">' + a.granted_by.substring(0, 16) + '\u2026</td>' +
'<td style="color:' + statusColor + ';font-size:10px;">' + status + '</td>';
abody.appendChild(tr);
}
}
}
// Join/Leave actions
window.joinPool = function() {
$('joinBtn').disabled = true;
fetch('/api/pool/join', { method: 'POST' })
.then(function(r) {
if (!r.ok) return r.json().then(function(j) { throw new Error(j.error || r.statusText); });
return r.json();
})
.then(function() { toast('Joined pool', 'success'); })
.catch(function(e) { toast('Join failed: ' + e.message, 'error'); })
.finally(function() { $('joinBtn').disabled = false; });
};
window.leavePool = function() {
$('leaveBtn').disabled = true;
fetch('/api/pool/leave', { method: 'POST' })
.then(function(r) {
if (!r.ok) return r.json().then(function(j) { throw new Error(j.error || r.statusText); });
return r.json();
})
.then(function() { toast('Left pool', 'success'); })
.catch(function(e) { toast('Leave failed: ' + e.message, 'error'); })
.finally(function() { $('leaveBtn').disabled = false; });
};
// SSE connection
var es = new EventSource('/events');
es.addEventListener('pool', function(e) {
try {
var snap = JSON.parse(e.data);
updatePool(snap);
} catch(err) { console.error('pool parse error', err); }
});
es.addEventListener('done', function() {
dot.className = 'status-dot done';
es.close();
});
es.onerror = function() { dot.className = 'status-dot disconnected'; };
es.onopen = function() { dot.className = 'status-dot'; };
})();
</script>
</body>
</html>
"##;

View file

@ -0,0 +1,274 @@
//! Pool coordinator actor — placement-aware CRUD facade.
//!
//! Owns an `Arc<Mutex<PoolDisseminator>>` for query access and delegates
//! storage operations to the co-located `DatastoreNode` actor.
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use swactor::actor::{ActorAddress, ActorInterface, Ctx};
use distribution::types::NodeId;
use shared_types::ContentHash;
use shared_types::pool::PoolConfig;
use crate::messages::{DatastoreNodeMsg, DatastoreResponse};
use super::disseminator::PoolDisseminator;
use super::messages::PoolCoordinatorMsg;
/// The pool coordinator actor.
pub struct PoolCoordinator {
node_id: NodeId,
pool_config: PoolConfig,
disseminator: Arc<Mutex<PoolDisseminator>>,
// Co-located actor addresses
datastore_addr: ActorAddress,
tick_count: u64,
}
impl PoolCoordinator {
pub fn new(
node_id: NodeId,
pool_config: PoolConfig,
disseminator: Arc<Mutex<PoolDisseminator>>,
datastore_addr: ActorAddress,
) -> Self {
Self {
node_id,
pool_config,
disseminator,
datastore_addr,
tick_count: 0,
}
}
/// Access the shared disseminator.
pub fn disseminator(&self) -> &Arc<Mutex<PoolDisseminator>> {
&self.disseminator
}
fn cluster_size(&self) -> usize {
let d = self.disseminator.lock().unwrap();
d.member_count().max(1)
}
// ─── Message handlers ──────────────────────────────────────────────
fn handle_pool_put(
&self,
ctx: &Ctx,
data: Vec<u8>,
name: Option<String>,
tags: BTreeMap<String, String>,
reply_to: ActorAddress,
) {
// Delegate to local DatastoreNode for now.
// Future: check capacity and redirect to best node.
let _ = ctx.send(
self.datastore_addr,
DatastoreNodeMsg::Put {
data,
name,
tags,
reply_to,
},
);
// Note: content announcement happens after PutOk is received.
// For now, caller is responsible for announcing content via PoolTick
// or a future PutOk callback.
}
fn handle_pool_get(
&self,
ctx: &Ctx,
content_hash: ContentHash,
reply_to: ActorAddress,
) {
// Try local first via DatastoreNode
let _ = ctx.send(
self.datastore_addr,
DatastoreNodeMsg::Get {
content_hash,
reply_to,
},
);
// Future: if local not found, use disseminator.locate_content()
// to fetch from a specific peer instead of fan-out.
}
fn handle_pool_delete(
&self,
ctx: &Ctx,
content_hash: ContentHash,
reply_to: ActorAddress,
) {
// Delete locally
let _ = ctx.send(
self.datastore_addr,
DatastoreNodeMsg::Delete {
content_hash,
reply_to,
},
);
// Announce tombstone via gossip
let cluster_size = self.cluster_size();
self.disseminator
.lock()
.unwrap()
.remove_content(content_hash, cluster_size);
}
fn handle_pool_list(
&self,
ctx: &Ctx,
name_filter: Option<String>,
reply_to: ActorAddress,
) {
let _ = ctx.send(
self.datastore_addr,
DatastoreNodeMsg::List {
name_filter,
all: false,
reply_to,
},
);
}
fn handle_pool_status(&self, ctx: &Ctx, reply_to: ActorAddress) {
let d = self.disseminator.lock().unwrap();
let (total_bytes, used_bytes) = d.pool_capacity_summary();
let members: Vec<String> = d
.active_members()
.iter()
.map(|id| id.0.iter().map(|b| format!("{b:02x}")).collect())
.collect();
let member_count = d.member_count();
let content_count = d.content_count();
drop(d);
let json = serde_json::json!({
"pool_name": self.pool_config.pool_name,
"pool_id": self.pool_config.pool_id.to_hex(),
"member_count": member_count,
"content_count": content_count,
"total_bytes": total_bytes,
"used_bytes": used_bytes,
"members": members,
});
let _ = ctx.send(
reply_to,
DatastoreResponse::PoolStatus {
json: json.to_string(),
},
);
}
fn handle_join_pool(&mut self, ctx: &Ctx, reply_to: ActorAddress) {
let cluster_size = self.cluster_size();
let mut d = self.disseminator.lock().unwrap();
if !d.is_node_authorized(&self.node_id) {
let _ = ctx.send(
reply_to,
DatastoreResponse::Error {
reason: "not authorized to join pool".into(),
},
);
return;
}
d.join(cluster_size);
d.announce_capacity(self.pool_config.capacity_bytes, 0, cluster_size);
drop(d);
let _ = ctx.send(reply_to, DatastoreResponse::Bool(true));
}
fn handle_leave_pool(&mut self, ctx: &Ctx, reply_to: ActorAddress) {
let cluster_size = self.cluster_size();
self.disseminator.lock().unwrap().leave(cluster_size);
let _ = ctx.send(reply_to, DatastoreResponse::Bool(true));
}
fn handle_grant_access(&mut self, ctx: &Ctx, target: NodeId, reply_to: ActorAddress) {
let cluster_size = self.cluster_size();
self.disseminator
.lock()
.unwrap()
.grant_access(target, cluster_size);
let _ = ctx.send(reply_to, DatastoreResponse::Bool(true));
}
fn handle_revoke_access(&mut self, ctx: &Ctx, target: NodeId, reply_to: ActorAddress) {
let cluster_size = self.cluster_size();
self.disseminator
.lock()
.unwrap()
.revoke_access(target, cluster_size);
let _ = ctx.send(reply_to, DatastoreResponse::Bool(true));
}
fn handle_pool_tick(&mut self) {
self.tick_count += 1;
// Periodic capacity re-announcement (every 100 ticks)
if self.tick_count % 100 == 0 {
let cluster_size = self.cluster_size();
self.disseminator.lock().unwrap().announce_capacity(
self.pool_config.capacity_bytes,
0, // TODO: query actual usage from BlobStore
cluster_size,
);
}
}
}
impl ActorInterface for PoolCoordinator {
type Incoming = PoolCoordinatorMsg;
type Response = DatastoreResponse;
fn handle(&mut self, ctx: &Ctx, msg: PoolCoordinatorMsg) {
match msg {
PoolCoordinatorMsg::PoolPut {
data,
name,
tags,
reply_to,
} => self.handle_pool_put(ctx, data, name, tags, reply_to),
PoolCoordinatorMsg::PoolGet {
content_hash,
reply_to,
} => self.handle_pool_get(ctx, content_hash, reply_to),
PoolCoordinatorMsg::PoolDelete {
content_hash,
reply_to,
} => self.handle_pool_delete(ctx, content_hash, reply_to),
PoolCoordinatorMsg::PoolList {
name_filter,
reply_to,
} => self.handle_pool_list(ctx, name_filter, reply_to),
PoolCoordinatorMsg::PoolStatus { reply_to } => {
self.handle_pool_status(ctx, reply_to)
}
PoolCoordinatorMsg::JoinPool { reply_to } => {
self.handle_join_pool(ctx, reply_to)
}
PoolCoordinatorMsg::LeavePool { reply_to } => {
self.handle_leave_pool(ctx, reply_to)
}
PoolCoordinatorMsg::GrantPoolAccess { target, reply_to } => {
self.handle_grant_access(ctx, target, reply_to)
}
PoolCoordinatorMsg::RevokePoolAccess { target, reply_to } => {
self.handle_revoke_access(ctx, target, reply_to)
}
PoolCoordinatorMsg::PoolTick => self.handle_pool_tick(),
}
}
}

View file

@ -0,0 +1,716 @@
//! Pool disseminator — gossip-converged state for pool membership,
//! capacity, content locations, and ACL.
//!
//! Implements `GossipChannel` to plug into the generic gossip system
//! via `DistributedNode::register_channel()`.
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use distribution::gossip_channel::{DisseminationBuffer, GossipChannel, deserialize_each, serialize_each};
use distribution::types::NodeId;
use shared_types::ContentHash;
use shared_types::pool::*;
// ─── PoolDisseminator ──────────────────────────────────────────────────────
/// Configuration for the pool disseminator.
#[derive(Debug, Clone)]
pub struct PoolDisseminatorConfig {
pub tombstone_ttl: u64,
pub gc_interval: u64,
}
impl Default for PoolDisseminatorConfig {
fn default() -> Self {
Self {
tombstone_ttl: 3600,
gc_interval: 1000,
}
}
}
/// Manages converged pool state via gossip dissemination.
#[derive(Debug)]
pub struct PoolDisseminator {
pool_id: PoolId,
pool_name: String,
local_node_id: NodeId,
// Converged state maps
members: HashMap<[u8; 32], PoolMemberEntry>,
capacity: HashMap<[u8; 32], PoolCapacityEntry>,
content_locations: HashMap<(ContentHash, [u8; 32]), ContentLocationEntry>,
acl: HashMap<[u8; 32], PoolACLEntry>,
// Dissemination buffer
buffer: DisseminationBuffer<PoolEntry>,
// Local generation counters
local_member_gen: u64,
local_capacity_gen: u64,
config: PoolDisseminatorConfig,
tick_count: u64,
}
impl PoolDisseminator {
pub fn new(pool_id: PoolId, pool_name: String, local_node_id: NodeId, lambda: usize) -> Self {
Self {
pool_id,
pool_name,
local_node_id,
members: HashMap::new(),
capacity: HashMap::new(),
content_locations: HashMap::new(),
acl: HashMap::new(),
buffer: DisseminationBuffer::new(lambda),
local_member_gen: 0,
local_capacity_gen: 0,
config: PoolDisseminatorConfig::default(),
tick_count: 0,
}
}
pub fn with_config(mut self, config: PoolDisseminatorConfig) -> Self {
self.config = config;
self
}
pub fn pool_id(&self) -> PoolId {
self.pool_id
}
// ─── Lifecycle methods ──────────────────────────────────────────────
/// Join the pool. Announces Active membership.
pub fn join(&mut self, cluster_size: usize) {
self.local_member_gen += 1;
let entry = PoolMemberEntry {
pool_id: self.pool_id,
node_id: self.local_node_id.0,
state: PoolMemberState::Active,
generation: self.local_member_gen,
};
self.merge_membership(entry.clone());
self.buffer.enqueue(PoolEntry::Membership(entry), cluster_size);
}
/// Leave the pool. Announces Left membership.
pub fn leave(&mut self, cluster_size: usize) {
self.local_member_gen += 1;
let entry = PoolMemberEntry {
pool_id: self.pool_id,
node_id: self.local_node_id.0,
state: PoolMemberState::Left,
generation: self.local_member_gen,
};
self.merge_membership(entry.clone());
self.buffer.enqueue(PoolEntry::Membership(entry), cluster_size);
}
/// Announce storage capacity.
pub fn announce_capacity(&mut self, total: u64, used: u64, cluster_size: usize) {
self.local_capacity_gen += 1;
let entry = PoolCapacityEntry {
pool_id: self.pool_id,
node_id: self.local_node_id.0,
total_bytes: total,
used_bytes: used,
generation: self.local_capacity_gen,
};
self.merge_capacity(entry.clone());
self.buffer.enqueue(PoolEntry::Capacity(entry), cluster_size);
}
/// Announce that this node has a piece of content.
pub fn announce_content(&mut self, hash: ContentHash, cluster_size: usize) {
let key = (hash, self.local_node_id.0);
let next_gen = self.content_locations.get(&key).map_or(1, |e| e.generation + 1);
let entry = ContentLocationEntry {
pool_id: self.pool_id,
content_hash: hash,
node_id: self.local_node_id.0,
generation: next_gen,
tombstone: false,
};
self.merge_content_location(entry.clone());
self.buffer.enqueue(PoolEntry::ContentLocation(entry), cluster_size);
}
/// Remove content announcement (tombstone).
pub fn remove_content(&mut self, hash: ContentHash, cluster_size: usize) {
let key = (hash, self.local_node_id.0);
let next_gen = self.content_locations.get(&key).map_or(1, |e| e.generation + 1);
let entry = ContentLocationEntry {
pool_id: self.pool_id,
content_hash: hash,
node_id: self.local_node_id.0,
generation: next_gen,
tombstone: true,
};
self.merge_content_location(entry.clone());
self.buffer.enqueue(PoolEntry::ContentLocation(entry), cluster_size);
}
/// Grant access to a node.
pub fn grant_access(&mut self, target: NodeId, cluster_size: usize) {
let next_gen = self.acl.get(&target.0).map_or(1, |e| e.generation + 1);
let entry = PoolACLEntry {
pool_id: self.pool_id,
node_id: target.0,
granted_by: self.local_node_id.0,
generation: next_gen,
revoked: false,
};
self.merge_acl(entry.clone());
self.buffer.enqueue(PoolEntry::ACL(entry), cluster_size);
}
/// Revoke access from a node.
pub fn revoke_access(&mut self, target: NodeId, cluster_size: usize) {
let next_gen = self.acl.get(&target.0).map_or(1, |e| e.generation + 1);
let entry = PoolACLEntry {
pool_id: self.pool_id,
node_id: target.0,
granted_by: self.local_node_id.0,
generation: next_gen,
revoked: true,
};
self.merge_acl(entry.clone());
self.buffer.enqueue(PoolEntry::ACL(entry), cluster_size);
}
// ─── Query API ──────────────────────────────────────────────────────
/// All active pool members.
pub fn active_members(&self) -> Vec<NodeId> {
self.members
.values()
.filter(|m| m.state == PoolMemberState::Active)
.map(|m| NodeId(m.node_id))
.collect()
}
/// Total and used capacity across the pool.
pub fn pool_capacity_summary(&self) -> (u64, u64) {
let mut total = 0u64;
let mut used = 0u64;
for cap in self.capacity.values() {
// Only count active members
if let Some(m) = self.members.get(&cap.node_id) {
if m.state == PoolMemberState::Active {
total = total.saturating_add(cap.total_bytes);
used = used.saturating_add(cap.used_bytes);
}
}
}
(total, used)
}
/// Find which nodes have a given content hash.
pub fn locate_content(&self, hash: &ContentHash) -> Vec<NodeId> {
self.content_locations
.iter()
.filter(|((h, _), entry)| h == hash && !entry.tombstone)
.map(|((_, node_id), _)| NodeId(*node_id))
.collect()
}
/// Find the node with the most free space.
pub fn node_with_most_free_space(&self) -> Option<NodeId> {
self.capacity
.values()
.filter(|cap| {
self.members
.get(&cap.node_id)
.is_some_and(|m| m.state == PoolMemberState::Active)
})
.max_by_key(|cap| cap.total_bytes.saturating_sub(cap.used_bytes))
.map(|cap| NodeId(cap.node_id))
}
/// Check if a node is authorized to join this pool.
pub fn is_node_authorized(&self, node_id: &NodeId) -> bool {
// If no ACL entries exist, the pool is open
if self.acl.is_empty() {
return true;
}
self.acl
.get(&node_id.0)
.is_some_and(|entry| !entry.revoked)
}
/// Number of active members.
pub fn member_count(&self) -> usize {
self.members
.values()
.filter(|m| m.state == PoolMemberState::Active)
.count()
}
/// Number of live content location entries (non-tombstone).
pub fn content_count(&self) -> usize {
self.content_locations
.values()
.filter(|e| !e.tombstone)
.count()
}
/// Serialize the current pool state to a JSON string for the dashboard.
pub fn snapshot_json(&self) -> String {
fn hex(bytes: &[u8; 32]) -> String {
bytes.iter().map(|b| format!("{b:02x}")).collect()
}
let (total_bytes, used_bytes) = self.pool_capacity_summary();
let members: Vec<serde_json::Value> = self
.members
.values()
.filter(|m| m.state == PoolMemberState::Active)
.map(|m| {
let cap = self.capacity.get(&m.node_id);
serde_json::json!({
"node_id": hex(&m.node_id),
"state": format!("{:?}", m.state),
"generation": m.generation,
"total_bytes": cap.map_or(0, |c| c.total_bytes),
"used_bytes": cap.map_or(0, |c| c.used_bytes),
})
})
.collect();
// Group content locations by hash
let mut by_hash: HashMap<ContentHash, Vec<[u8; 32]>> = HashMap::new();
for ((hash, _), entry) in &self.content_locations {
if !entry.tombstone {
by_hash.entry(*hash).or_default().push(entry.node_id);
}
}
let content_locations: Vec<serde_json::Value> = by_hash
.iter()
.map(|(hash, nodes)| {
serde_json::json!({
"content_hash": hash.to_hex(),
"nodes": nodes.iter().map(hex).collect::<Vec<_>>(),
"replica_count": nodes.len(),
})
})
.collect();
let acl: Vec<serde_json::Value> = self
.acl
.values()
.map(|a| {
serde_json::json!({
"node_id": hex(&a.node_id),
"granted_by": hex(&a.granted_by),
"revoked": a.revoked,
})
})
.collect();
let acl_mode = if self.acl.is_empty() { "open" } else { "allow-list" };
serde_json::json!({
"pool_name": self.pool_name,
"pool_id": self.pool_id.to_hex(),
"member_count": self.member_count(),
"content_count": self.content_count(),
"total_bytes": total_bytes,
"used_bytes": used_bytes,
"members": members,
"content_locations": content_locations,
"acl": acl,
"acl_mode": acl_mode,
})
.to_string()
}
// ─── Internal merge logic ───────────────────────────────────────────
fn merge_membership(&mut self, entry: PoolMemberEntry) -> bool {
let key = entry.node_id;
if let Some(existing) = self.members.get(&key) {
if entry.generation <= existing.generation {
return false;
}
}
self.members.insert(key, entry);
true
}
fn merge_capacity(&mut self, entry: PoolCapacityEntry) -> bool {
let key = entry.node_id;
if let Some(existing) = self.capacity.get(&key) {
if entry.generation <= existing.generation {
return false;
}
}
self.capacity.insert(key, entry);
true
}
fn merge_content_location(&mut self, entry: ContentLocationEntry) -> bool {
let key = (entry.content_hash, entry.node_id);
if let Some(existing) = self.content_locations.get(&key) {
if entry.generation <= existing.generation {
return false;
}
}
self.content_locations.insert(key, entry);
true
}
fn merge_acl(&mut self, entry: PoolACLEntry) -> bool {
let key = entry.node_id;
if let Some(existing) = self.acl.get(&key) {
if entry.generation <= existing.generation {
return false;
}
}
self.acl.insert(key, entry);
true
}
/// Merge a single pool entry and return whether state changed.
fn merge_entry(&mut self, entry: PoolEntry) -> bool {
match entry {
PoolEntry::Membership(m) => self.merge_membership(m),
PoolEntry::Capacity(c) => self.merge_capacity(c),
PoolEntry::ContentLocation(cl) => self.merge_content_location(cl),
PoolEntry::ACL(a) => self.merge_acl(a),
}
}
/// Take pending entries (internal, typed).
fn take_pending_inner(&mut self, max_count: usize) -> Vec<PoolEntry> {
self.buffer.take(max_count)
}
/// Apply incoming entries (internal, typed).
fn apply_incoming_inner(&mut self, entries: Vec<PoolEntry>, cluster_size: usize) {
for entry in entries {
if self.merge_entry(entry.clone()) {
self.buffer.enqueue(entry, cluster_size);
}
}
}
/// Re-enqueue all state (internal).
fn re_disseminate_all_inner(&mut self, cluster_size: usize) {
let mut all_entries: Vec<PoolEntry> = Vec::new();
for m in self.members.values().cloned() {
all_entries.push(PoolEntry::Membership(m));
}
for c in self.capacity.values().cloned() {
all_entries.push(PoolEntry::Capacity(c));
}
for cl in self.content_locations.values().cloned() {
all_entries.push(PoolEntry::ContentLocation(cl));
}
for a in self.acl.values().cloned() {
all_entries.push(PoolEntry::ACL(a));
}
self.buffer.re_enqueue_all(all_entries, cluster_size);
}
/// GC: evict tombstones past TTL.
fn gc_tick_inner(&mut self) {
self.tick_count += 1;
if self.tick_count % self.config.gc_interval != 0 {
return;
}
let ttl = self.config.tombstone_ttl;
let tick = self.tick_count;
// GC left members
self.members.retain(|_, m| {
if m.state == PoolMemberState::Left {
m.generation + ttl > tick
} else {
true
}
});
// GC tombstoned content locations
self.content_locations.retain(|_, cl| {
if cl.tombstone {
cl.generation + ttl > tick
} else {
true
}
});
// GC revoked ACL entries
self.acl.retain(|_, a| {
if a.revoked {
a.generation + ttl > tick
} else {
true
}
});
}
}
// ─── SharedPoolChannel ─────────────────────────────────────────────────────
/// Wrapper around `Arc<Mutex<PoolDisseminator>>` that implements `GossipChannel`.
///
/// This enables shared ownership between the `PoolCoordinator` actor
/// (which needs query/lifecycle access) and `DistributedNode` (which
/// drives gossip piggyback).
pub struct SharedPoolChannel {
inner: Arc<Mutex<PoolDisseminator>>,
}
impl SharedPoolChannel {
pub fn new(disseminator: Arc<Mutex<PoolDisseminator>>) -> Self {
Self { inner: disseminator }
}
/// Consume the channel and return the underlying `Arc<Mutex<PoolDisseminator>>`.
pub fn into_inner(self) -> Arc<Mutex<PoolDisseminator>> {
self.inner
}
}
impl GossipChannel for SharedPoolChannel {
fn topic_tag(&self) -> &'static str {
"pool"
}
fn take_pending_bytes(&mut self, max_entries: usize) -> Vec<Vec<u8>> {
let entries = self.inner.lock().unwrap().take_pending_inner(max_entries);
serialize_each(&entries)
}
fn apply_incoming_bytes(&mut self, entries: &[Vec<u8>], cluster_size: usize) {
let parsed: Vec<PoolEntry> = deserialize_each(entries);
self.inner.lock().unwrap().apply_incoming_inner(parsed, cluster_size);
}
fn re_disseminate_all(&mut self, cluster_size: usize) {
self.inner.lock().unwrap().re_disseminate_all_inner(cluster_size);
}
fn on_node_death(&mut self, _node_id: &NodeId) {
// Pool membership is explicit (join/leave), not auto-removed on node death.
// Capacity becomes unreliable but we don't remove it.
}
fn gc_tick(&mut self) {
self.inner.lock().unwrap().gc_tick_inner();
}
}
#[cfg(test)]
mod tests {
use super::*;
fn node_id(b: u8) -> NodeId {
NodeId([b; 32])
}
fn make_disseminator(b: u8) -> PoolDisseminator {
PoolDisseminator::new(
PoolId::from_name("test-pool"),
"test-pool".into(),
node_id(b),
3,
)
}
#[test]
fn join_and_query_members() {
let mut d = make_disseminator(1);
d.join(2);
assert_eq!(d.member_count(), 1);
assert_eq!(d.active_members(), vec![node_id(1)]);
}
#[test]
fn leave_removes_from_active() {
let mut d = make_disseminator(1);
d.join(2);
d.leave(2);
assert_eq!(d.member_count(), 0);
assert!(d.active_members().is_empty());
}
#[test]
fn announce_and_locate_content() {
let mut d = make_disseminator(1);
d.join(2);
let hash = ContentHash::of(b"test-data");
d.announce_content(hash, 2);
let locations = d.locate_content(&hash);
assert_eq!(locations, vec![node_id(1)]);
}
#[test]
fn remove_content_tombstones() {
let mut d = make_disseminator(1);
d.join(2);
let hash = ContentHash::of(b"test-data");
d.announce_content(hash, 2);
d.remove_content(hash, 2);
assert!(d.locate_content(&hash).is_empty());
}
#[test]
fn capacity_summary() {
let mut d = make_disseminator(1);
d.join(2);
d.announce_capacity(1000, 300, 2);
let (total, used) = d.pool_capacity_summary();
assert_eq!(total, 1000);
assert_eq!(used, 300);
}
#[test]
fn acl_grant_and_check() {
let mut d = make_disseminator(1);
d.grant_access(node_id(2), 2);
assert!(d.is_node_authorized(&node_id(2)));
assert!(!d.is_node_authorized(&node_id(3)));
}
#[test]
fn acl_revoke() {
let mut d = make_disseminator(1);
d.grant_access(node_id(2), 2);
d.revoke_access(node_id(2), 2);
assert!(!d.is_node_authorized(&node_id(2)));
}
#[test]
fn empty_acl_means_open() {
let d = make_disseminator(1);
assert!(d.is_node_authorized(&node_id(99)));
}
#[test]
fn higher_generation_wins_merge() {
let mut d1 = make_disseminator(1);
let mut d2 = make_disseminator(2);
// d1 joins
d1.join(2);
// d1 leaves
d1.leave(2);
// Gossip d1's entries to d2 out of order:
// First send the Active (gen 1), then the Left (gen 2)
let active_entry = PoolEntry::Membership(PoolMemberEntry {
pool_id: PoolId::from_name("test-pool"),
node_id: [1u8; 32],
state: PoolMemberState::Active,
generation: 1,
});
let left_entry = PoolEntry::Membership(PoolMemberEntry {
pool_id: PoolId::from_name("test-pool"),
node_id: [1u8; 32],
state: PoolMemberState::Left,
generation: 2,
});
// Apply Left first (gen 2), then Active (gen 1) — Active should be rejected
d2.merge_entry(left_entry);
let changed = d2.merge_entry(active_entry);
assert!(!changed, "lower generation should not win");
// d2 should see node_id(1) as Left
assert_eq!(d2.member_count(), 0); // Active count is 0
}
#[test]
fn two_disseminators_converge_via_gossip_exchange() {
let mut d1 = make_disseminator(1);
let mut d2 = make_disseminator(2);
// d1 joins and announces content
d1.join(2);
let hash = ContentHash::of(b"shared-file");
d1.announce_content(hash, 2);
// d2 joins
d2.join(2);
// Simulate gossip: d1 → d2
let pending = d1.take_pending_inner(100);
let bytes = serialize_each(&pending);
let parsed: Vec<PoolEntry> = deserialize_each(&bytes);
d2.apply_incoming_inner(parsed, 2);
// d2 should now see d1 as a member and know about the content
assert_eq!(d2.member_count(), 2);
assert_eq!(d2.locate_content(&hash), vec![node_id(1)]);
// Simulate gossip: d2 → d1
let pending = d2.take_pending_inner(100);
let bytes = serialize_each(&pending);
let parsed: Vec<PoolEntry> = deserialize_each(&bytes);
d1.apply_incoming_inner(parsed, 2);
// d1 should now see d2 as a member
assert_eq!(d1.member_count(), 2);
}
#[test]
fn three_node_convergence_loop() {
let pool = PoolId::from_name("test-pool");
let mut nodes: Vec<PoolDisseminator> = (0..3)
.map(|i| PoolDisseminator::new(pool, "test-pool".into(), node_id(i as u8), 3))
.collect();
// Each node joins
for n in &mut nodes {
n.join(3);
}
// Node 0 announces content
let hash = ContentHash::of(b"convergence-test");
nodes[0].announce_content(hash, 3);
// Run 5 gossip rounds where each node exchanges with all others
for _ in 0..5 {
// Collect pending from each node
let pending_bytes: Vec<Vec<Vec<u8>>> = nodes
.iter_mut()
.map(|n| serialize_each(&n.take_pending_inner(100)))
.collect();
// Apply each node's pending to all other nodes
for (sender_idx, bytes) in pending_bytes.iter().enumerate() {
for (receiver_idx, node) in nodes.iter_mut().enumerate() {
if sender_idx != receiver_idx {
let parsed: Vec<PoolEntry> = deserialize_each(bytes);
node.apply_incoming_inner(parsed, 3);
}
}
}
}
// All nodes should agree on membership and content locations
for (i, node) in nodes.iter().enumerate() {
assert_eq!(node.member_count(), 3, "node {i} should see 3 members");
assert_eq!(
node.locate_content(&hash),
vec![node_id(0)],
"node {i} should know content is on node 0"
);
}
}
}

View file

@ -0,0 +1,79 @@
//! Messages for the `PoolCoordinator` actor.
use std::collections::BTreeMap;
use distribution::types::NodeId;
use shared_types::ContentHash;
/// Messages handled by the `PoolCoordinator` actor.
#[derive(Debug, Clone)]
pub enum PoolCoordinatorMsg {
// ── User-facing operations ──────────────────────────────────────────
/// Store data in the pool (placement-aware).
PoolPut {
data: Vec<u8>,
name: Option<String>,
tags: BTreeMap<String, String>,
reply_to: swactor::actor::ActorAddress,
},
/// Retrieve data from the pool (location-aware).
PoolGet {
content_hash: ContentHash,
reply_to: swactor::actor::ActorAddress,
},
/// Delete data from the pool.
PoolDelete {
content_hash: ContentHash,
reply_to: swactor::actor::ActorAddress,
},
/// List objects in the pool.
PoolList {
name_filter: Option<String>,
reply_to: swactor::actor::ActorAddress,
},
/// Pool status (members, capacity, content count).
PoolStatus {
reply_to: swactor::actor::ActorAddress,
},
// ── Pool lifecycle ──────────────────────────────────────────────────
/// Join the pool.
JoinPool {
reply_to: swactor::actor::ActorAddress,
},
/// Leave the pool.
LeavePool {
reply_to: swactor::actor::ActorAddress,
},
// ── Auth management ─────────────────────────────────────────────────
/// Grant a node access to the pool.
GrantPoolAccess {
target: NodeId,
reply_to: swactor::actor::ActorAddress,
},
/// Revoke a node's access to the pool.
RevokePoolAccess {
target: NodeId,
reply_to: swactor::actor::ActorAddress,
},
// ── Periodic ────────────────────────────────────────────────────────
/// Periodic tick: announce capacity, drive dissemination.
PoolTick,
}
/// Pool-specific response variants.
#[derive(Debug, Clone)]
pub enum PoolResponse {
/// Pool status snapshot.
PoolStatus {
pool_name: String,
pool_id_hex: String,
member_count: usize,
content_count: usize,
total_bytes: u64,
used_bytes: u64,
members: Vec<String>,
},
}

View file

@ -0,0 +1,8 @@
//! Pooled datastore protocol.
//!
//! A shared storage pool where multiple nodes contribute storage capacity
//! and converge on a shared view of what content lives where.
pub mod disseminator;
pub mod messages;
pub mod coordinator;

View file

@ -48,6 +48,7 @@ fn dashboard_reflects_datastore_operations() {
port: dash_port, port: dash_port,
..Default::default() ..Default::default()
}); });
dash.start_http_standalone();
let collector = dashboard::collector::StatsCollector::new(2); let collector = dashboard::collector::StatsCollector::new(2);
let mut rt = Runtime::new(RuntimeConfig { let mut rt = Runtime::new(RuntimeConfig {

View file

@ -0,0 +1,341 @@
//! Pool protocol integration tests.
//!
//! Tests the pool disseminator's convergence behavior, the gossip channel
//! integration, and the coordinator actor's lifecycle.
use std::sync::{Arc, Mutex};
use distribution::gossip_channel::{GossipChannel, serialize_each};
use distribution::types::NodeId;
use shared_types::ContentHash;
use shared_types::pool::*;
use swactor_datastore::pool::disseminator::{PoolDisseminator, SharedPoolChannel};
fn node_id(b: u8) -> NodeId {
NodeId([b; 32])
}
fn make_pool_disseminator(node_byte: u8) -> PoolDisseminator {
PoolDisseminator::new(
PoolId::from_name("test-pool"),
"test-pool".into(),
node_id(node_byte),
3,
)
}
// ─── Disseminator convergence scenarios ────────────────────────────────────
/// Two nodes join a pool and exchange gossip until they converge
/// on identical state.
#[test]
fn two_nodes_converge_on_membership() {
let mut d1 = make_pool_disseminator(1);
let mut d2 = make_pool_disseminator(2);
d1.join(2);
d2.join(2);
// Simulate 3 gossip rounds
for _ in 0..3 {
let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1)));
let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2)));
let bytes1 = ch1.take_pending_bytes(100);
let bytes2 = ch2.take_pending_bytes(100);
ch1.apply_incoming_bytes(&bytes2, 2);
ch2.apply_incoming_bytes(&bytes1, 2);
d1 = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap();
d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap();
}
assert_eq!(d1.member_count(), 2);
assert_eq!(d2.member_count(), 2);
}
/// Content announced on one node is visible on another after gossip.
#[test]
fn content_location_propagates_via_gossip() {
let mut d1 = make_pool_disseminator(1);
let mut d2 = make_pool_disseminator(2);
let hash = ContentHash::of(b"test-content");
d1.join(2);
d2.join(2);
d1.announce_content(hash, 2);
// Gossip d1 → d2
let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1)));
let bytes = ch1.take_pending_bytes(100);
let _d1 = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap();
let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2)));
ch2.apply_incoming_bytes(&bytes, 2);
let d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap();
assert_eq!(d2.locate_content(&hash), vec![node_id(1)]);
}
/// A node's leave is propagated and removes it from the active member list
/// on the receiving node.
#[test]
fn leave_propagates_via_gossip() {
let mut d1 = make_pool_disseminator(1);
let mut d2 = make_pool_disseminator(2);
d1.join(2);
d2.join(2);
// Exchange so both see each other
let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1)));
let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2)));
let bytes1 = ch1.take_pending_bytes(100);
let bytes2 = ch2.take_pending_bytes(100);
ch1.apply_incoming_bytes(&bytes2, 2);
ch2.apply_incoming_bytes(&bytes1, 2);
d1 = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap();
d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap();
assert_eq!(d1.member_count(), 2);
assert_eq!(d2.member_count(), 2);
// d1 leaves
d1.leave(2);
// Gossip leave to d2
let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1)));
let bytes = ch1.take_pending_bytes(100);
let _d1 = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap();
let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2)));
ch2.apply_incoming_bytes(&bytes, 2);
let d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap();
assert_eq!(d2.member_count(), 1); // only d2 remains active
}
/// Content tombstone propagates and removes the location.
#[test]
fn content_deletion_propagates() {
let mut d1 = make_pool_disseminator(1);
let mut d2 = make_pool_disseminator(2);
let hash = ContentHash::of(b"ephemeral");
d1.join(2);
d2.join(2);
d1.announce_content(hash, 2);
// Propagate content announcement
let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1)));
let bytes = ch1.take_pending_bytes(100);
d1 = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap();
let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2)));
ch2.apply_incoming_bytes(&bytes, 2);
d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap();
assert_eq!(d2.locate_content(&hash).len(), 1);
// d1 removes content
d1.remove_content(hash, 2);
// Propagate tombstone
let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1)));
let bytes = ch1.take_pending_bytes(100);
let _ = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap();
let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2)));
ch2.apply_incoming_bytes(&bytes, 2);
d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap();
assert!(d2.locate_content(&hash).is_empty());
}
/// ACL grant propagates to other nodes.
#[test]
fn acl_grant_propagates() {
let mut d1 = make_pool_disseminator(1);
let mut d2 = make_pool_disseminator(2);
d1.grant_access(node_id(3), 2);
let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1)));
let bytes = ch1.take_pending_bytes(100);
let _ = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap();
let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2)));
ch2.apply_incoming_bytes(&bytes, 2);
d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap();
assert!(d2.is_node_authorized(&node_id(3)));
assert!(!d2.is_node_authorized(&node_id(4)));
}
/// Capacity announcement propagates and is reflected in summary.
#[test]
fn capacity_propagates_and_summarizes() {
let mut d1 = make_pool_disseminator(1);
let mut d2 = make_pool_disseminator(2);
d1.join(2);
d2.join(2);
d1.announce_capacity(1_000_000, 100_000, 2);
d2.announce_capacity(2_000_000, 200_000, 2);
// Exchange
let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1)));
let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2)));
let bytes1 = ch1.take_pending_bytes(100);
let bytes2 = ch2.take_pending_bytes(100);
ch1.apply_incoming_bytes(&bytes2, 2);
ch2.apply_incoming_bytes(&bytes1, 2);
d1 = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap();
d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap();
let (total1, used1) = d1.pool_capacity_summary();
let (total2, used2) = d2.pool_capacity_summary();
assert_eq!(total1, 3_000_000);
assert_eq!(used1, 300_000);
assert_eq!(total2, 3_000_000);
assert_eq!(used2, 300_000);
}
/// The node_with_most_free_space query returns the correct node.
#[test]
fn placement_query_picks_node_with_most_space() {
let mut d = make_pool_disseminator(1);
d.join(3);
// Simulate node 2 joining and having lots of space
let member2 = PoolEntry::Membership(PoolMemberEntry {
pool_id: PoolId::from_name("test-pool"),
node_id: [2u8; 32],
state: PoolMemberState::Active,
generation: 1,
});
let cap2 = PoolEntry::Capacity(PoolCapacityEntry {
pool_id: PoolId::from_name("test-pool"),
node_id: [2u8; 32],
total_bytes: 10_000_000,
used_bytes: 1_000_000,
generation: 1,
});
// Simulate node 3 with less free space
let member3 = PoolEntry::Membership(PoolMemberEntry {
pool_id: PoolId::from_name("test-pool"),
node_id: [3u8; 32],
state: PoolMemberState::Active,
generation: 1,
});
let cap3 = PoolEntry::Capacity(PoolCapacityEntry {
pool_id: PoolId::from_name("test-pool"),
node_id: [3u8; 32],
total_bytes: 5_000_000,
used_bytes: 4_000_000,
generation: 1,
});
let entries = vec![member2, cap2, member3, cap3];
let bytes = serialize_each(&entries);
let mut ch = SharedPoolChannel::new(Arc::new(Mutex::new(d)));
ch.apply_incoming_bytes(&bytes, 3);
d = Arc::try_unwrap(ch.into_inner()).unwrap().into_inner().unwrap();
// d1 has no capacity announced, node 2 has 9M free, node 3 has 1M free
let best = d.node_with_most_free_space().unwrap();
assert_eq!(best, node_id(2));
}
/// Five-node convergence: all nodes join, one announces content, everyone converges.
#[test]
fn five_node_pool_converges() {
let pool = PoolId::from_name("five-pool");
let mut nodes: Vec<PoolDisseminator> = (0..5)
.map(|i| PoolDisseminator::new(pool, "five-pool".into(), node_id(i as u8), 3))
.collect();
// All join
for n in &mut nodes {
n.join(5);
}
// Node 0 and 2 announce content
let hash_a = ContentHash::of(b"file-a");
let hash_b = ContentHash::of(b"file-b");
nodes[0].announce_content(hash_a, 5);
nodes[2].announce_content(hash_b, 5);
// Run 10 gossip rounds
for _ in 0..10 {
let mut channels: Vec<SharedPoolChannel> = nodes
.into_iter()
.map(|d| SharedPoolChannel::new(Arc::new(Mutex::new(d))))
.collect();
let pending: Vec<Vec<Vec<u8>>> = channels
.iter_mut()
.map(|ch| ch.take_pending_bytes(100))
.collect();
for (recv_idx, ch) in channels.iter_mut().enumerate() {
for (send_idx, bytes) in pending.iter().enumerate() {
if recv_idx != send_idx {
ch.apply_incoming_bytes(bytes, 5);
}
}
}
nodes = channels
.into_iter()
.map(|ch| Arc::try_unwrap(ch.into_inner()).unwrap().into_inner().unwrap())
.collect();
}
for (i, node) in nodes.iter().enumerate() {
assert_eq!(node.member_count(), 5, "node {i} should see 5 members");
assert_eq!(
node.locate_content(&hash_a),
vec![node_id(0)],
"node {i} should locate file-a on node 0"
);
assert_eq!(
node.locate_content(&hash_b),
vec![node_id(2)],
"node {i} should locate file-b on node 2"
);
}
}
// ─── GossipChannel wire format tests ───────────────────────────────────────
/// Verify the GossipChannel topic tag is correct.
#[test]
fn shared_pool_channel_topic_tag() {
let d = make_pool_disseminator(1);
let ch = SharedPoolChannel::new(Arc::new(Mutex::new(d)));
assert_eq!(GossipChannel::topic_tag(&ch), "pool");
}
/// Verify serialization round-trip through GossipChannel bytes interface.
#[test]
fn gossip_channel_bytes_roundtrip() {
let mut d = make_pool_disseminator(1);
d.join(2);
let hash = ContentHash::of(b"roundtrip");
d.announce_content(hash, 2);
let mut ch = SharedPoolChannel::new(Arc::new(Mutex::new(d)));
let bytes = ch.take_pending_bytes(100);
assert!(!bytes.is_empty());
// Apply to a fresh disseminator
let d2 = make_pool_disseminator(2);
let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2)));
ch2.apply_incoming_bytes(&bytes, 2);
let d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap();
assert_eq!(d2.member_count(), 1);
assert_eq!(d2.locate_content(&hash), vec![node_id(1)]);
}

View file

@ -0,0 +1,268 @@
//! Generic gossip channel abstraction.
//!
//! Provides `GossipChannel` — a trait for any topic that wants to piggyback
//! on SWIM protocol messages — and `DisseminationBuffer<T>` — a reusable
//! budget-limited dissemination queue that replaces the 4 independent copies
//! of the `Λ * ceil(log₂(n))` pattern.
use serde::{Serialize, de::DeserializeOwned};
// ─── GossipChannel trait ────────────────────────────────────────────────────
/// A gossip channel that can piggyback serialized entries on SWIM messages.
///
/// Each channel has a unique topic tag and handles its own serialization.
/// The wire transport uses type-erased `Vec<u8>` entries.
pub trait GossipChannel: Send {
/// Unique string tag identifying this channel in the piggyback payload.
fn topic_tag(&self) -> &'static str;
/// Take up to `max_entries` pending entries, serialized as bytes.
fn take_pending_bytes(&mut self, max_entries: usize) -> Vec<Vec<u8>>;
/// Apply incoming entries (deserialized from bytes) received from gossip.
fn apply_incoming_bytes(&mut self, entries: &[Vec<u8>], cluster_size: usize);
/// Re-enqueue all state for dissemination (anti-entropy on membership recovery).
fn re_disseminate_all(&mut self, cluster_size: usize);
/// Handle a node being declared dead.
fn on_node_death(&mut self, node_id: &crate::types::NodeId);
/// Periodic garbage collection tick.
fn gc_tick(&mut self);
}
// ─── DisseminationBuffer<T> ────────────────────────────────────────────────
/// A queued entry with a remaining transmit budget.
#[derive(Debug, Clone)]
struct BufferEntry<T> {
item: T,
remaining: usize,
}
/// Reusable generic dissemination buffer.
///
/// Manages budget-limited gossip dissemination for any entry type.
/// Each entry is transmitted `Λ * ceil(log₂(n))` times before eviction.
#[derive(Debug)]
pub struct DisseminationBuffer<T> {
entries: Vec<BufferEntry<T>>,
lambda: usize,
}
impl<T: Clone> DisseminationBuffer<T> {
/// Create a new buffer with the given dissemination multiplier (Λ).
pub fn new(lambda: usize) -> Self {
Self {
entries: Vec::new(),
lambda,
}
}
/// Compute the transmit budget: `Λ * ceil(log₂(max(n, 2)))`.
pub fn transmit_budget(&self, cluster_size: usize) -> usize {
let n = cluster_size.max(2) as f64;
let log_n = n.log2().ceil() as usize;
self.lambda * log_n.max(1)
}
/// Enqueue an entry for dissemination. Does not check for duplicates.
pub fn enqueue(&mut self, item: T, cluster_size: usize) {
let budget = self.transmit_budget(cluster_size);
self.entries.push(BufferEntry {
item,
remaining: budget,
});
}
/// Enqueue an entry, replacing an existing one if `matcher` returns true.
/// If no match is found, pushes a new entry.
pub fn enqueue_or_replace<F>(&mut self, item: T, cluster_size: usize, matcher: F)
where
F: Fn(&T) -> bool,
{
let budget = self.transmit_budget(cluster_size);
if let Some(existing) = self.entries.iter_mut().find(|e| matcher(&e.item)) {
existing.item = item;
existing.remaining = budget;
return;
}
self.entries.push(BufferEntry {
item,
remaining: budget,
});
}
/// Take up to `max_count` entries for piggyback.
/// Decrements remaining budget and evicts exhausted entries.
pub fn take(&mut self, max_count: usize) -> Vec<T> {
let count = max_count.min(self.entries.len());
let mut result = Vec::with_capacity(count);
for entry in self.entries.iter_mut().take(count) {
result.push(entry.item.clone());
entry.remaining = entry.remaining.saturating_sub(1);
}
self.entries.retain(|e| e.remaining > 0);
result
}
/// Re-enqueue all given items with fresh budgets.
pub fn re_enqueue_all(&mut self, items: impl IntoIterator<Item = T>, cluster_size: usize) {
for item in items {
self.enqueue(item, cluster_size);
}
}
/// Retain only entries matching the predicate.
pub fn retain<F>(&mut self, mut predicate: F)
where
F: FnMut(&T) -> bool,
{
self.entries.retain(|e| predicate(&e.item));
}
/// Number of queued entries.
pub fn len(&self) -> usize {
self.entries.len()
}
/// Whether the buffer is empty.
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
}
// ─── Serialization helpers ─────────────────────────────────────────────────
/// Serialize a slice of items to a Vec of byte vectors.
pub fn serialize_each<T: Serialize>(items: &[T]) -> Vec<Vec<u8>> {
items
.iter()
.filter_map(|item| serde_json::to_vec(item).ok())
.collect()
}
/// Deserialize a slice of byte vectors into items, skipping failures.
pub fn deserialize_each<T: DeserializeOwned>(entries: &[Vec<u8>]) -> Vec<T> {
entries
.iter()
.filter_map(|bytes| serde_json::from_slice(bytes).ok())
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn budget_math_two_nodes() {
let buf: DisseminationBuffer<u32> = DisseminationBuffer::new(3);
// log2(2) = 1, ceil = 1, 3 * 1 = 3
assert_eq!(buf.transmit_budget(2), 3);
}
#[test]
fn budget_math_single_node_floors_to_two() {
let buf: DisseminationBuffer<u32> = DisseminationBuffer::new(3);
// cluster_size=1 → max(1,2)=2, log2(2)=1, 3*1=3
assert_eq!(buf.transmit_budget(1), 3);
}
#[test]
fn budget_math_32_nodes() {
let buf: DisseminationBuffer<u32> = DisseminationBuffer::new(3);
// log2(32) = 5, 3 * 5 = 15
assert_eq!(buf.transmit_budget(32), 15);
}
#[test]
fn budget_math_33_nodes() {
let buf: DisseminationBuffer<u32> = DisseminationBuffer::new(3);
// log2(33) ≈ 5.04, ceil = 6, 3 * 6 = 18
assert_eq!(buf.transmit_budget(33), 18);
}
#[test]
fn enqueue_take_evicts_after_budget() {
let mut buf = DisseminationBuffer::new(3);
buf.enqueue(42u32, 2); // budget = 3
// Take 3 times — each take decrements once
let r1 = buf.take(1);
assert_eq!(r1, vec![42]);
assert_eq!(buf.len(), 1);
let r2 = buf.take(1);
assert_eq!(r2, vec![42]);
assert_eq!(buf.len(), 1);
let r3 = buf.take(1);
assert_eq!(r3, vec![42]);
// After 3 takes, budget exhausted → evicted
assert_eq!(buf.len(), 0);
}
#[test]
fn enqueue_or_replace_replaces_matching_entry() {
let mut buf = DisseminationBuffer::new(3);
buf.enqueue_or_replace(("key", 1), 2, |e| e.0 == "key");
buf.enqueue_or_replace(("key", 2), 2, |e| e.0 == "key");
assert_eq!(buf.len(), 1);
let taken = buf.take(1);
assert_eq!(taken, vec![("key", 2)]);
}
#[test]
fn enqueue_or_replace_adds_when_no_match() {
let mut buf = DisseminationBuffer::new(3);
buf.enqueue_or_replace(("a", 1), 2, |e| e.0 == "a");
buf.enqueue_or_replace(("b", 2), 2, |e| e.0 == "b");
assert_eq!(buf.len(), 2);
}
#[test]
fn re_enqueue_all_refreshes_budgets() {
let mut buf = DisseminationBuffer::new(3);
buf.enqueue(1u32, 2);
buf.enqueue(2u32, 2);
// Drain them
for _ in 0..3 {
buf.take(2);
}
assert!(buf.is_empty());
// Re-enqueue
buf.re_enqueue_all(vec![1, 2, 3], 2);
assert_eq!(buf.len(), 3);
}
#[test]
fn retain_removes_non_matching() {
let mut buf = DisseminationBuffer::new(3);
buf.enqueue(1u32, 2);
buf.enqueue(2u32, 2);
buf.enqueue(3u32, 2);
buf.retain(|item| *item != 2);
assert_eq!(buf.len(), 2);
let taken = buf.take(3);
assert_eq!(taken, vec![1, 3]);
}
#[test]
fn serialize_deserialize_roundtrip() {
let items = vec![1u32, 2, 3];
let bytes = serialize_each(&items);
let recovered: Vec<u32> = deserialize_each(&bytes);
assert_eq!(recovered, items);
}
}

View file

@ -507,15 +507,31 @@ impl IrohDriver {
self.connections.remove(&node_id); self.connections.remove(&node_id);
} }
// Resolve relay URL: explicit cache → SWIM metadata gossip → own home relay
let relay = self.peer_relay_urls.get(&node_id).cloned()
.or_else(|| {
self.node
.relay_url(&node_id)
.and_then(|s| s.parse::<iroh::RelayUrl>().ok())
})
.or_else(|| self.endpoint.addr().relay_urls().next().cloned());
let endpoint = self.endpoint.clone(); let endpoint = self.endpoint.clone();
let conn = if let Some(relay) = self.peer_relay_urls.get(&node_id) { let connect_timeout = Duration::from_secs(2);
let addr = EndpointAddr::new(key).with_relay_url(relay.clone()); let conn = if let Some(relay) = relay {
let addr = EndpointAddr::new(key).with_relay_url(relay);
self.rt.block_on(async { self.rt.block_on(async {
endpoint.connect(addr, ALPN).await match tokio::time::timeout(connect_timeout, endpoint.connect(addr, ALPN)).await {
Ok(result) => result.map_err(|e| -> Box<dyn std::error::Error> { Box::new(e) }),
Err(_) => Err("connect timeout".into()),
}
})? })?
} else { } else {
self.rt.block_on(async { self.rt.block_on(async {
endpoint.connect(key, ALPN).await match tokio::time::timeout(connect_timeout, endpoint.connect(key, ALPN)).await {
Ok(result) => result.map_err(|e| -> Box<dyn std::error::Error> { Box::new(e) }),
Err(_) => Err("connect timeout".into()),
}
})? })?
}; };

View file

@ -0,0 +1,252 @@
//! Shared types for the pooled datastore protocol.
//!
//! Types live in `shared-types` because both `distribution` and `datastore`
//! depend on them.
use std::fmt;
use serde::{Deserialize, Serialize};
use crate::ContentHash;
// Re-export NodeId-shaped bytes — pool uses the same 32-byte node identifier.
// The actual NodeId type lives in `distribution::types`, but we use raw [u8; 32]
// here to avoid a circular dependency. Callers convert as needed.
// ─── PoolId ────────────────────────────────────────────────────────────────
/// Unique pool identifier: blake3(name_bytes).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct PoolId(pub [u8; 32]);
impl PoolId {
/// Create a pool ID from a human-readable name.
pub fn from_name(name: &str) -> Self {
PoolId(*blake3::hash(name.as_bytes()).as_bytes())
}
/// Encode as lowercase hex string.
pub fn to_hex(&self) -> String {
let mut s = String::with_capacity(64);
for b in &self.0 {
use fmt::Write;
write!(s, "{:02x}", b).unwrap();
}
s
}
/// Parse a 64-character hex string into a PoolId.
pub fn from_hex(hex: &str) -> Option<Self> {
if hex.len() != 64 {
return None;
}
let mut bytes = [0u8; 32];
for (i, chunk) in hex.as_bytes().chunks(2).enumerate() {
let hi = hex_digit(chunk[0])?;
let lo = hex_digit(chunk[1])?;
bytes[i] = (hi << 4) | lo;
}
Some(PoolId(bytes))
}
}
impl fmt::Display for PoolId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
for b in &self.0[..8] {
write!(f, "{:02x}", b)?;
}
write!(f, "\u{2026}")
}
}
// ─── Pool Membership ───────────────────────────────────────────────────────
/// Whether a node is actively participating in a pool.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum PoolMemberState {
Active,
Left,
}
/// A node's membership in a pool. Higher generation always wins.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PoolMemberEntry {
pub pool_id: PoolId,
pub node_id: [u8; 32],
pub state: PoolMemberState,
pub generation: u64,
}
// ─── Pool Capacity ─────────────────────────────────────────────────────────
/// A node's storage capacity announcement. Higher generation always wins.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PoolCapacityEntry {
pub pool_id: PoolId,
pub node_id: [u8; 32],
pub total_bytes: u64,
pub used_bytes: u64,
pub generation: u64,
}
// ─── Content Location ──────────────────────────────────────────────────────
/// Where a content hash is stored. Key: (pool_id, content_hash, node_id).
/// Higher generation wins. Tombstones indicate deletion.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ContentLocationEntry {
pub pool_id: PoolId,
pub content_hash: ContentHash,
pub node_id: [u8; 32],
pub generation: u64,
pub tombstone: bool,
}
// ─── Pool ACL ──────────────────────────────────────────────────────────────
/// Authorization for a node to join a pool. Higher generation wins.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PoolACLEntry {
pub pool_id: PoolId,
pub node_id: [u8; 32],
pub granted_by: [u8; 32],
pub generation: u64,
pub revoked: bool,
}
// ─── Tagged Union ──────────────────────────────────────────────────────────
/// All pool entry variants, used for gossip serialization.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum PoolEntry {
Membership(PoolMemberEntry),
Capacity(PoolCapacityEntry),
ContentLocation(ContentLocationEntry),
ACL(PoolACLEntry),
}
// ─── Pool Config ───────────────────────────────────────────────────────────
/// Configuration for a pool.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PoolConfig {
pub pool_name: String,
pub pool_id: PoolId,
/// Total bytes this node pledges to the pool.
pub capacity_bytes: u64,
/// Tombstone TTL in ticks before GC.
pub tombstone_ttl: u64,
/// GC interval in ticks.
pub gc_interval: u64,
/// Dissemination multiplier (Λ).
pub dissemination_lambda: usize,
}
impl PoolConfig {
pub fn new(pool_name: &str, capacity_bytes: u64) -> Self {
Self {
pool_name: pool_name.to_string(),
pool_id: PoolId::from_name(pool_name),
capacity_bytes,
tombstone_ttl: 3600,
gc_interval: 1000,
dissemination_lambda: 3,
}
}
}
// ─── Helpers ───────────────────────────────────────────────────────────────
fn hex_digit(b: u8) -> Option<u8> {
match b {
b'0'..=b'9' => Some(b - b'0'),
b'a'..=b'f' => Some(b - b'a' + 10),
b'A'..=b'F' => Some(b - b'A' + 10),
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn pool_id_from_name_is_deterministic() {
let a = PoolId::from_name("test-pool");
let b = PoolId::from_name("test-pool");
assert_eq!(a, b);
}
#[test]
fn pool_id_different_names_differ() {
let a = PoolId::from_name("pool-a");
let b = PoolId::from_name("pool-b");
assert_ne!(a, b);
}
#[test]
fn pool_id_hex_roundtrip() {
let id = PoolId::from_name("roundtrip-test");
let hex = id.to_hex();
let recovered = PoolId::from_hex(&hex).unwrap();
assert_eq!(id, recovered);
}
#[test]
fn pool_entry_serde_roundtrip() {
let entry = PoolEntry::Membership(PoolMemberEntry {
pool_id: PoolId::from_name("test"),
node_id: [1u8; 32],
state: PoolMemberState::Active,
generation: 1,
});
let bytes = serde_json::to_vec(&entry).unwrap();
let recovered: PoolEntry = serde_json::from_slice(&bytes).unwrap();
match recovered {
PoolEntry::Membership(m) => {
assert_eq!(m.node_id, [1u8; 32]);
assert_eq!(m.state, PoolMemberState::Active);
assert_eq!(m.generation, 1);
}
_ => panic!("wrong variant"),
}
}
#[test]
fn higher_generation_wins_for_membership() {
let old = PoolMemberEntry {
pool_id: PoolId::from_name("test"),
node_id: [1u8; 32],
state: PoolMemberState::Active,
generation: 1,
};
let new = PoolMemberEntry {
pool_id: PoolId::from_name("test"),
node_id: [1u8; 32],
state: PoolMemberState::Left,
generation: 2,
};
assert!(new.generation > old.generation);
}
#[test]
fn content_location_tombstone_semantics() {
let live = ContentLocationEntry {
pool_id: PoolId::from_name("test"),
content_hash: ContentHash::of(b"hello"),
node_id: [1u8; 32],
generation: 1,
tombstone: false,
};
let dead = ContentLocationEntry {
pool_id: PoolId::from_name("test"),
content_hash: ContentHash::of(b"hello"),
node_id: [1u8; 32],
generation: 2,
tombstone: true,
};
assert!(!live.tombstone);
assert!(dead.tombstone);
assert!(dead.generation > live.generation);
}
}

View file

@ -789,8 +789,16 @@ fn run_iroh(
let seed_bytes = parse_node_id_str(&seed_str).expect("invalid seed node ID (expected hex or base58)"); let seed_bytes = parse_node_id_str(&seed_str).expect("invalid seed node ID (expected hex or base58)");
let seed_key = let seed_key =
iroh::PublicKey::from_bytes(&seed_bytes).expect("invalid seed public key"); iroh::PublicKey::from_bytes(&seed_bytes).expect("invalid seed public key");
let mut seed_addr = iroh::EndpointAddr::from(seed_key);
// Include relay URLs so iroh can locate the seed through the relay
for host in &relay_hosts {
let url_str = format!("http://{host}:{relay_port}/");
if let Ok(url) = url_str.parse::<iroh::RelayUrl>() {
seed_addr = seed_addr.with_relay_url(url);
}
}
eprintln!("Joining cluster via seed {}", base58_encode(&seed_bytes)); eprintln!("Joining cluster via seed {}", base58_encode(&seed_bytes));
driver.join(&[iroh::EndpointAddr::from(seed_key)]); driver.join(&[seed_addr]);
} }
// Spawn and register actors // Spawn and register actors

View file

@ -0,0 +1,423 @@
# Pooled Datastore & Sim-Cluster — Development History
> Adds a gossip-converged pooled storage protocol, a generic gossip channel
> abstraction, a pool dashboard page, shared pool types, iroh connection
> hardening, and a Docker-free multi-process sim-cluster test harness.
>
> ~18 new/modified files · ~2,400 insertions
>
> *Branch: `pooled-datastore`*
---
## Table of Contents
1. [Overview & Motivation](#1-overview--motivation)
2. [What Was Built](#2-what-was-built)
3. [Pooled Storage Protocol](#3-pooled-storage-protocol)
4. [Generic Gossip Channel Abstraction](#4-generic-gossip-channel-abstraction)
5. [Pool Disseminator](#5-pool-disseminator)
6. [Pool Coordinator Actor](#6-pool-coordinator-actor)
7. [Shared Pool Types](#7-shared-pool-types)
8. [Dashboard Pool Page](#8-dashboard-pool-page)
9. [Iroh Connection Hardening](#9-iroh-connection-hardening)
10. [Sim-Cluster Test Harness](#10-sim-cluster-test-harness)
11. [Design Decisions & Tradeoffs](#11-design-decisions--tradeoffs)
12. [Test Coverage](#12-test-coverage)
13. [Known Gaps & Future Work](#13-known-gaps--future-work)
---
## 1. Overview & Motivation
The existing datastore provides content-addressed storage on individual nodes,
but there is no mechanism for multiple nodes to form a shared storage pool —
knowing who has what content, how much capacity each node offers, or where to
place new data.
This branch introduces a **pooled datastore protocol** that layers on top of
the existing content-addressed datastore. Multiple nodes join a named pool,
gossip their membership/capacity/content-locations via SWIM piggyback, and
converge on a shared view of the pool's state. This enables:
- **Content location**: find which node(s) hold a given content hash without
fan-out queries.
- **Capacity-aware placement**: route new writes to the node with the most free
space.
- **Pool ACL**: optional allow-list to restrict which nodes can join a pool.
- **Live observability**: a new dashboard page shows pool membership, capacity
bars, content location map, and ACL state in real time via SSE.
Separately, the branch also introduces:
- A **generic gossip channel abstraction** (`GossipChannel` trait +
`DisseminationBuffer<T>`) that replaces the 4 duplicated dissemination
patterns in the distribution crate.
- A **sim-cluster test harness** (`cargo xtask test sim-cluster` / `cargo
xtask sim-cluster`) that spawns real multi-process clusters with a local iroh
relay — no Docker required.
- **Iroh connection hardening**: relay URL resolution cascade and connect
timeouts to prevent indefinite hangs during peer connection.
---
## 2. What Was Built
| Component | Crate / Location | Lines |
|-----------|-----------------|-------|
| Pool types (PoolId, entries, config) | `crates/shared-types/src/pool.rs` | ~170 |
| Gossip channel trait + DisseminationBuffer | `crates/distribution/src/gossip_channel.rs` | ~270 |
| Pool disseminator (CRDT state + gossip) | `crates/datastore/src/pool/disseminator.rs` | ~720 |
| Pool coordinator actor | `crates/datastore/src/pool/coordinator.rs` | ~275 |
| Pool messages | `crates/datastore/src/pool/messages.rs` | ~80 |
| Pool dashboard HTML/JS | `crates/dashboard/src/pool_html.rs` | ~390 |
| Sim-cluster harness | `xtask/src/sim_cluster.rs` | ~700 |
| Pool integration tests | `crates/datastore/tests/pool_tests.rs` | ~340 |
| Docker compose (dev cluster) | `tests/docker/docker-compose.dev-cluster.yml` | ~75 |
Modified files:
| File | Change |
|------|--------|
| `crates/distribution/src/iroh_driver.rs` | Relay URL cascade + connect timeout |
| `crates/swactor-node/src/main.rs` | Seed node relay URL hints for iroh |
| `crates/datastore/tests/dashboard_integration_test.rs` | Start HTTP standalone |
| `xtask/src/main.rs` | `sim-cluster` subcommand + test group |
| `xtask/Cargo.toml` | reqwest, tokio, iroh-relay deps |
---
## 3. Pooled Storage Protocol
The pool protocol is a set of four CRDT entry types that converge via gossip:
```
┌────────────────────────────────────────────────────┐
│ Pool State (per node) │
├──────────────┬─────────────┬───────────┬───────────┤
│ Membership │ Capacity │ Content │ ACL │
│ │ │ Location │ │
│ node→state │ node→bytes │ (hash, │ node→ │
│ (Active/Left)│ (total/used)│ node)→ │ grant/ │
│ │ │ tombstone│ revoke │
├──────────────┴─────────────┴───────────┴───────────┤
│ Higher generation always wins │
│ (last-writer-wins register per key) │
└────────────────────────────────────────────────────┘
```
**Convergence rule**: For each entry type, the key is derived from the entry
(e.g. `node_id` for membership, `(content_hash, node_id)` for content
locations). When two entries share a key, the one with the higher `generation`
wins. This makes all merges commutative, associative, and idempotent — a CRDT.
**Deletion**: Content locations and ACL entries use tombstones (`tombstone:
true` / `revoked: true`) with a generation bump. Tombstones are garbage
collected after a configurable TTL.
**Dissemination**: All entries go through a shared `DisseminationBuffer<PoolEntry>`
which transmits each entry `Λ * ceil(log₂(n))` times before eviction, matching
the standard SWIM protocol budget.
---
## 4. Generic Gossip Channel Abstraction
**File**: `crates/distribution/src/gossip_channel.rs`
Before this branch, SWIM piggyback dissemination was hardcoded for membership
updates, directory entries, and dead-letter notifications — each with its own
copy of the `Λ * ceil(log₂(n))` budget logic.
The new abstraction provides:
- **`GossipChannel` trait**: A topic-tagged channel that produces/consumes
`Vec<u8>` entries for piggyback. Methods: `topic_tag()`,
`take_pending_bytes()`, `apply_incoming_bytes()`, `re_disseminate_all()`,
`on_node_death()`, `gc_tick()`.
- **`DisseminationBuffer<T>`**: A generic budget-limited queue. Entries are
enqueued with a transmit budget of `Λ * ceil(log₂(n))` and evicted after
exhaustion. Supports `enqueue`, `enqueue_or_replace` (idempotent upsert),
`take`, `re_enqueue_all`, and `retain`.
- **Serialization helpers**: `serialize_each()` and `deserialize_each()` for
converting between typed entries and `Vec<u8>`.
The pool disseminator is the first consumer, plugging into the distribution
layer via `SharedPoolChannel` which wraps `Arc<Mutex<PoolDisseminator>>`.
---
## 5. Pool Disseminator
**File**: `crates/datastore/src/pool/disseminator.rs`
The core state machine. Manages four `HashMap` tables (membership, capacity,
content locations, ACL) and a single `DisseminationBuffer<PoolEntry>`.
Key methods:
- **Lifecycle**: `join()`, `leave()` — announce membership state changes.
- **Storage**: `announce_content()`, `remove_content()`, `announce_capacity()`.
- **ACL**: `grant_access()`, `revoke_access()`, `is_node_authorized()`.
- **Queries**: `active_members()`, `member_count()`, `content_count()`,
`locate_content()`, `node_with_most_free_space()`, `pool_capacity_summary()`.
- **Dashboard**: `snapshot_json()` — full JSON snapshot for SSE.
- **Internal**: `merge_entry()` applies the higher-generation-wins rule.
`take_pending_inner()` / `apply_incoming_inner()` drive gossip exchange.
`SharedPoolChannel` wraps this in an `Arc<Mutex<>>` and implements
`GossipChannel`, bridging ownership between the `PoolCoordinator` actor
(lifecycle/queries) and `DistributedNode` (gossip transport).
---
## 6. Pool Coordinator Actor
**File**: `crates/datastore/src/pool/coordinator.rs`
An actor implementing `ActorInterface` for `PoolCoordinatorMsg`. It acts as a
placement-aware CRUD facade:
- **PoolPut/Get/Delete/List**: Delegates to the co-located `DatastoreNode`
actor. Future: redirect to best node based on capacity.
- **PoolStatus**: Queries the disseminator and returns a JSON status snapshot.
- **JoinPool/LeavePool**: Checks ACL authorization, then calls the
disseminator.
- **GrantPoolAccess/RevokePoolAccess**: Manages the allow-list.
- **PoolTick**: Periodic capacity re-announcement (every 100 ticks).
---
## 7. Shared Pool Types
**File**: `crates/shared-types/src/pool.rs`
Types live in `shared-types` to avoid circular dependencies between
`distribution` and `datastore`:
- **`PoolId`**: `blake3(name_bytes)` — 32-byte deterministic pool identifier.
Supports hex encoding/decoding and truncated display.
- **`PoolMemberEntry`**: Node membership with `Active`/`Left` state.
- **`PoolCapacityEntry`**: Storage capacity announcement (total/used bytes).
- **`ContentLocationEntry`**: Where a content hash is stored, with tombstone
support.
- **`PoolACLEntry`**: Authorization grant/revoke with `granted_by` provenance.
- **`PoolEntry`**: Tagged enum wrapping all four entry types for gossip
serialization.
- **`PoolConfig`**: Pool configuration (name, capacity, TTL, GC interval, Λ).
---
## 8. Dashboard Pool Page
**File**: `crates/dashboard/src/pool_html.rs`
A new `/pool` page in the dashboard with:
- **Summary cards**: pool name, member count, content count, total/used
capacity.
- **Capacity bars**: per-node usage with color thresholds (green < 70%, orange
< 90%, red >= 90%).
- **Members table**: node ID (truncated with tooltip), state, total/used/free.
- **Content location map**: content hash → replica count → node list.
- **ACL panel**: open mode indicator or allow-list table.
- **Join/Leave buttons**: POST to `/api/pool/join` and `/api/pool/leave`.
- **Live updates**: SSE `pool` events drive real-time state refresh.
---
## 9. Iroh Connection Hardening
**File**: `crates/distribution/src/iroh_driver.rs`
Two problems fixed:
1. **Relay URL resolution cascade**: When connecting to a peer, the driver now
tries three sources in order: (a) explicit relay URL cache from prior
connections, (b) SWIM metadata gossip (via `node.relay_url()`), (c) the
local node's own home relay. Previously only the explicit cache was checked,
causing connections to fail when the cache was empty.
2. **Connect timeout**: All `endpoint.connect()` calls now have a 2-second
`tokio::time::timeout` wrapper. Previously, connections could hang
indefinitely if a peer was unreachable.
**File**: `crates/swactor-node/src/main.rs`
Seed node addresses now include relay URLs so iroh can locate the seed through
the relay server, rather than relying solely on direct addressing.
---
## 10. Sim-Cluster Test Harness
**File**: `xtask/src/sim_cluster.rs`
A new test stage and development tool that spawns real multi-process swactor
clusters without Docker:
### Test mode: `cargo xtask test sim-cluster`
Runs 4 scenarios sequentially, each with a fresh 5-node cluster:
| # | Scenario | Validates |
|---|----------|-----------|
| 1 | Cluster convergence | All 5 nodes see >= 4 alive peers, routing table >= 4 |
| 2 | Node death detection | Kill node 2, survivors detect alive drop, dead count >= 1 |
| 3 | Killed node rejoins | Kill node 2, restart it, rejoined node sees alive >= 1 |
| 4 | Actors resolvable | Each node has >= 2 directory entries, total >= 10 |
### Interactive mode: `cargo xtask sim-cluster --nodes N`
Spawns a persistent cluster for development. Prints dashboard URLs and blocks
until Ctrl-C.
### Infrastructure
- **Local relay server**: Embedded `iroh-relay` server on an ephemeral port.
Nodes connect through the relay rather than requiring direct connectivity.
- **RAII lifecycle**: `SimCluster` owns child processes and SIGTERM's them on
drop. `RelayServer` owns its tokio runtime.
- **Config generation**: Each node gets a `node.toml` with dashboard port,
actor count, relay host/port, and optional seed node ID.
- **Seed key discovery**: Polls the seed node's key file on disk to extract the
public key before spawning joiner nodes.
- **HTTP observation**: Polls `/api/distribution` on each node's dashboard.
Uses `serde_json::Value` to avoid compile-time coupling to protocol types.
- **Node lifecycle**: `kill_node()` sends SIGTERM, `restart_node()` re-spawns
with the same config (non-seed nodes get the seed's public key).
### Dev cluster compose
**File**: `tests/docker/docker-compose.dev-cluster.yml`
A 3-node Docker Compose file for development with pool configuration
(`--pool-name dev-pool --pool-capacity 104857600`). Uses a bridge network with
static IPs.
---
## 11. Design Decisions & Tradeoffs
**Higher-generation-wins CRDT over vector clocks**: Pool entries use a simple
monotonic generation counter per entry key. This is sufficient because each
entry has a single writer (the node that owns it). Vector clocks would add
complexity without benefit since there are no concurrent writers for the same
key.
**Tombstones with TTL over immediate deletion**: Content locations and ACL
revocations use tombstones that propagate via gossip before being GC'd. Without
tombstones, a deleted entry could be re-introduced by a node that hasn't yet
received the deletion.
**Shared `Arc<Mutex<>>` over message-passing for disseminator**: The pool
disseminator needs to be accessed by both the coordinator actor (for
lifecycle/queries) and the distribution layer (for gossip). Rather than adding
an actor-to-actor message protocol, the disseminator is wrapped in
`Arc<Mutex<PoolDisseminator>>`. The lock is held only briefly for individual
operations.
**Sim-cluster over Docker for testing**: Docker adds build time, image
management, and network configuration complexity. The sim-cluster spawns bare
processes on localhost, uses an embedded iroh relay, and tears down in
milliseconds. Scenarios that previously required Docker Compose now run with
`cargo xtask test sim-cluster`.
**HTTP polling over direct protocol observation**: The sim-cluster observes
node state via HTTP (`/api/distribution`) rather than linking against protocol
types. This makes the test harness resilient to protocol changes and mirrors
how an operator would observe a real cluster.
**Pool types in `shared-types`**: Pool entry types live in `shared-types`
rather than `datastore` to avoid a circular dependency — `distribution` needs
to know about pool entries for gossip serialization, and `datastore` depends on
`distribution`.
---
## 12. Test Coverage
### Unit tests (disseminator internals)
In `crates/datastore/src/pool/disseminator.rs`:
- `join_and_query_members` — join lifecycle
- `leave_removes_from_active` — leave lifecycle
- `announce_and_locate_content` — content announcement + query
- `remove_content_tombstones` — tombstone semantics
- `capacity_summary` — capacity aggregation
- `acl_grant_and_check` / `acl_revoke` / `empty_acl_means_open` — ACL logic
- `higher_generation_wins_merge` — CRDT merge rule
- `two_disseminators_converge_via_gossip_exchange` — two-node gossip
- `three_node_convergence_loop` — multi-round gossip convergence
### Unit tests (gossip channel)
In `crates/distribution/src/gossip_channel.rs`:
- `budget_math_*` (4 tests) — transmit budget calculation
- `enqueue_take_evicts_after_budget` — budget exhaustion
- `enqueue_or_replace_*` (2 tests) — idempotent upsert
- `re_enqueue_all_refreshes_budgets` — anti-entropy
- `retain_removes_non_matching` — predicate-based eviction
- `serialize_deserialize_roundtrip` — wire format
### Unit tests (shared types)
In `crates/shared-types/src/pool.rs`:
- `pool_id_from_name_is_deterministic` / `pool_id_different_names_differ`
- `pool_id_hex_roundtrip`
- `pool_entry_serde_roundtrip`
- `higher_generation_wins_for_membership`
- `content_location_tombstone_semantics`
### Integration tests (pool protocol)
In `crates/datastore/tests/pool_tests.rs`:
- `two_nodes_converge_on_membership` — two-node gossip convergence
- `content_location_propagates_via_gossip` — cross-node content discovery
- `leave_propagates_via_gossip` — membership leave propagation
- `content_deletion_propagates` — tombstone propagation
- `acl_grant_propagates` — ACL gossip
- `capacity_propagates_and_summarizes` — capacity gossip + aggregation
- `placement_query_picks_node_with_most_space` — capacity-aware placement
- `five_node_pool_converges` — 5-node full convergence
- `shared_pool_channel_topic_tag` — GossipChannel interface
- `gossip_channel_bytes_roundtrip` — wire format through GossipChannel
### Sim-cluster scenarios (multi-process)
In `xtask/src/sim_cluster.rs`:
- Cluster convergence (5 nodes)
- Node death detection (kill + observe)
- Killed node rejoins (kill + restart + observe)
- Actors resolvable (directory entry propagation)
---
## 13. Known Gaps & Future Work
- **Remote content fetch**: `PoolGet` currently only checks the local
datastore. It should use `locate_content()` to fetch from the node that
actually has the content.
- **Capacity-aware placement**: `PoolPut` delegates to the local datastore.
It should use `node_with_most_free_space()` to route writes to the best node.
- **Actual usage tracking**: `PoolTick` re-announces capacity with `used: 0`.
It should query the `BlobStore` for actual disk usage.
- **GossipChannel integration**: The `GossipChannel` trait and
`SharedPoolChannel` are built but not yet wired into `DistributedNode`'s
piggyback system. The existing hardcoded dissemination channels need to be
migrated to the new trait.
- **Dashboard SSE integration**: The pool dashboard HTML is built, but the
server-side SSE event source for `pool` events needs to be wired to the
`PoolDisseminator::snapshot_json()` method.
- **Sim-cluster namespace isolation**: The harness uses high ephemeral ports
for isolation. Full Linux network namespace isolation (as designed in
`TEST_ISOLATION.md`) is a future enhancement.
- **Sim-cluster in CI**: The sim-cluster test group is opt-in and excluded
from `essential`/`all`. Once proven stable, it should be added to CI.

View file

@ -0,0 +1,77 @@
services:
seed:
image: swactor-dev-cluster
command:
- "--identity-dir"
- "/identity"
- "--listen"
- "0.0.0.0:11204"
- "--no-relay"
- "--pool-name"
- "dev-pool"
- "--pool-capacity"
- "104857600"
- "--actors"
- "2"
volumes:
- /home/aaron/swactor-pooled-datastore/.dev-cluster/seed-identity:/identity:ro
networks:
dev-cluster:
ipv4_address: 10.0.2.10
ports:
- "9100:9090"
node-2:
image: swactor-dev-cluster
command:
- "--seed-node-id"
- "d5469553d4408073022fee3e41c99cf9778bf3621e430f75f8c8ae7d6de0153f"
- "--seed-addrs"
- "10.0.2.10:11204"
- "--listen"
- "0.0.0.0:11204"
- "--no-relay"
- "--pool-name"
- "dev-pool"
- "--pool-capacity"
- "104857600"
- "--actors"
- "2"
networks:
dev-cluster:
ipv4_address: 10.0.2.11
ports:
- "9101:9090"
depends_on:
- seed
node-3:
image: swactor-dev-cluster
command:
- "--seed-node-id"
- "d5469553d4408073022fee3e41c99cf9778bf3621e430f75f8c8ae7d6de0153f"
- "--seed-addrs"
- "10.0.2.10:11204"
- "--listen"
- "0.0.0.0:11204"
- "--no-relay"
- "--pool-name"
- "dev-pool"
- "--pool-capacity"
- "104857600"
- "--actors"
- "2"
networks:
dev-cluster:
ipv4_address: 10.0.2.12
ports:
- "9102:9090"
depends_on:
- seed
networks:
dev-cluster:
driver: bridge
ipam:
config:
- subnet: 10.0.2.0/24

View file

@ -9,3 +9,6 @@ toml = "0.8"
serde = { version = "1", features = ["derive"] } serde = { version = "1", features = ["derive"] }
serde_json = "1" serde_json = "1"
libc = "0.2" libc = "0.2"
reqwest = { version = "0.12", features = ["blocking", "json"] }
tokio = { version = "1", features = ["rt-multi-thread"] }
iroh-relay = { version = "0.96", features = ["server"] }

View file

@ -1,4 +1,5 @@
mod deploy; mod deploy;
mod sim_cluster;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::process::Command; use std::process::Command;
@ -107,6 +108,13 @@ enum Cmd {
dirs: Vec<String>, dirs: Vec<String>,
}, },
/// Launch a local sim-cluster (relay + N nodes) for development
SimCluster {
/// Number of nodes (default: 5)
#[arg(long, default_value = "5")]
nodes: usize,
},
/// Deploy swactor to remote machines /// Deploy swactor to remote machines
Deploy { Deploy {
/// Deploy via Docker over SSH (build image, push, run containers) /// Deploy via Docker over SSH (build image, push, run containers)
@ -285,6 +293,7 @@ TEST GROUPS:
distribution Distribution protocol + datastore distribution Distribution protocol + datastore
cluster-sims Deterministic cluster simulations cluster-sims Deterministic cluster simulations
integrated HTTP API + dashboard end-to-end tests integrated HTTP API + dashboard end-to-end tests
sim-cluster Multi-process cluster with local iroh relay
essential core + distribution + integrated (merge gate) essential core + distribution + integrated (merge gate)
all Every test group all Every test group
@ -317,6 +326,7 @@ fn print_list() {
println!(); println!();
} }
println!(" {:<14}Multi-process cluster with local iroh relay", "sim-cluster");
println!(" {:<14}core + distribution + integrated (merge gate)", "essential"); println!(" {:<14}core + distribution + integrated (merge gate)", "essential");
println!(" {:<14}Every test group", "all"); println!(" {:<14}Every test group", "all");
} }
@ -337,6 +347,11 @@ fn run_test(group: Option<String>, list: bool) {
} }
}; };
if group_name == "sim-cluster" {
sim_cluster::run();
return;
}
let groups = match groups_for(&group_name) { let groups = match groups_for(&group_name) {
Some(g) => g, Some(g) => g,
None => { None => {
@ -930,6 +945,7 @@ fn main() {
} => run_cli(url, key, extra, &config.cli), } => run_cli(url, key, extra, &config.cli),
Cmd::InitNode { role, dir } => run_init_node(&role, dir.as_deref()), Cmd::InitNode { role, dir } => run_init_node(&role, dir.as_deref()),
Cmd::GenPeers { dirs } => run_gen_peers(&dirs), Cmd::GenPeers { dirs } => run_gen_peers(&dirs),
Cmd::SimCluster { nodes } => sim_cluster::run_interactive(nodes),
Cmd::Deploy { docker, config, skip_build, skip_verify, skip_peers } => { Cmd::Deploy { docker, config, skip_build, skip_verify, skip_peers } => {
let config = config.unwrap_or_else(|| { let config = config.unwrap_or_else(|| {
if docker { ".deploy/docker.toml" } else { ".deploy/deploy.toml" }.into() if docker { ".deploy/docker.toml" } else { ".deploy/deploy.toml" }.into()

701
xtask/src/sim_cluster.rs Normal file
View file

@ -0,0 +1,701 @@
//! sim-cluster: multi-process cluster tests using iroh transport.
//!
//! Spawns 5 swactor nodes with iroh transport connected through a local
//! relay server, then runs the same scenarios as `tests/docker/tests/cluster.rs`
//! without Docker.
use std::fs;
use std::net::Ipv4Addr;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::thread;
use std::time::{Duration, Instant};
use crate::workspace_root;
// ── Constants ───────────────────────────────────────────────────────────
const NODE_COUNT: usize = 5;
const ACTORS_PER_NODE: usize = 2;
const TIMEOUT: Duration = Duration::from_secs(30);
/// Dashboard HTTP ports — high ports to avoid conflicts.
fn dashboard_port(index: usize) -> u16 {
19091 + index as u16
}
fn all_dashboard_ports(node_count: usize) -> Vec<u16> {
(0..node_count).map(dashboard_port).collect()
}
// ── Local relay server ──────────────────────────────────────────────────
/// Owns a tokio runtime + iroh-relay server. RAII cleanup on drop.
struct RelayServer {
_server: iroh_relay::server::Server,
_rt: tokio::runtime::Runtime,
port: u16,
}
impl RelayServer {
fn start() -> Self {
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(1)
.enable_all()
.build()
.expect("failed to create tokio runtime for relay");
let server = rt.block_on(async {
iroh_relay::server::Server::spawn(iroh_relay::server::ServerConfig::<(), ()> {
relay: Some(iroh_relay::server::RelayConfig {
http_bind_addr: (Ipv4Addr::LOCALHOST, 0).into(),
tls: None,
limits: Default::default(),
key_cache_capacity: Some(256),
access: iroh_relay::server::AccessConfig::Everyone,
}),
quic: None,
metrics_addr: None,
})
.await
})
.expect("failed to spawn relay server");
let addr = server.http_addr().expect("relay has no HTTP address");
let port = addr.port();
RelayServer {
_server: server,
_rt: rt,
port,
}
}
}
// ── Build ───────────────────────────────────────────────────────────────
fn build_swactor(root: &Path) -> PathBuf {
let status = Command::new("cargo")
.args(["build", "-p", "swactor-node"])
.current_dir(root)
.status()
.expect("failed to run cargo build");
if !status.success() {
eprintln!("cargo build failed");
std::process::exit(1);
}
let binary = root.join("target/debug/swactor");
if !binary.exists() {
eprintln!("binary not found at {}", binary.display());
std::process::exit(1);
}
binary
}
// ── Config generation ───────────────────────────────────────────────────
fn write_node_config(
dir: &Path,
index: usize,
relay_port: u16,
seed_public_key: Option<&str>,
) -> PathBuf {
let identity_dir = dir.join("identity");
fs::create_dir_all(&identity_dir).expect("failed to create identity dir");
let config_path = dir.join("node.toml");
let dashboard = dashboard_port(index);
let mut config = format!(
r#"dashboard_port = {dashboard}
actors = {ACTORS_PER_NODE}
no_datastore = true
identity_dir = "{identity}"
relay = false
relay_port = {relay_port}
relay_hosts = ["127.0.0.1"]
"#,
identity = identity_dir.display(),
);
if let Some(seed_id) = seed_public_key {
config.push_str(&format!("seed_node_id = \"{seed_id}\"\n"));
}
fs::write(&config_path, &config).expect("failed to write node config");
config_path
}
// ── Seed key discovery ──────────────────────────────────────────────────
fn read_seed_public_key(run_dir: &Path) -> String {
let key_path = run_dir.join("node-0/identity/node.key.json");
let deadline = Instant::now() + Duration::from_secs(10);
loop {
if Instant::now() > deadline {
panic!(
"timed out waiting for seed key file: {}",
key_path.display()
);
}
if key_path.exists() {
if let Ok(data) = fs::read_to_string(&key_path) {
if let Ok(json) = serde_json::from_str::<serde_json::Value>(&data) {
if let Some(pk) = json.get("public_key").and_then(|v| v.as_str()) {
return pk.to_string();
}
}
}
}
thread::sleep(Duration::from_millis(100));
}
}
// ── Dashboard readiness ─────────────────────────────────────────────────
fn wait_for_dashboard(port: u16, timeout: Duration) {
let start = Instant::now();
loop {
if start.elapsed() > timeout {
panic!("timed out waiting for dashboard on port {port}");
}
if poll_distribution(port).is_some() {
return;
}
thread::sleep(Duration::from_millis(200));
}
}
// ── Cluster handle (RAII) ───────────────────────────────────────────────
struct SimCluster {
children: Vec<(usize, Child)>,
run_dir: PathBuf,
binary: PathBuf,
relay: RelayServer,
seed_public_key: String,
_node_count: usize,
}
impl SimCluster {
fn spawn(binary: &Path, run_dir: &Path, node_count: usize) -> Self {
fs::create_dir_all(run_dir).expect("failed to create run dir");
// Phase 0: start relay
let relay = RelayServer::start();
println!(
" relay at http://127.0.0.1:{}/ (port {})",
relay.port, relay.port
);
let mut children = Vec::new();
// Phase 1: spawn seed node (index 0) — no seed_node_id
let child = spawn_node(binary, run_dir, 0, relay.port, None);
children.push((0, child));
// Phase 2: wait for seed's key file
let seed_public_key = read_seed_public_key(run_dir);
println!(" seed key: {seed_public_key}");
// Wait for seed's dashboard to be ready before spawning joiners
wait_for_dashboard(dashboard_port(0), Duration::from_secs(15));
// Phase 3: spawn remaining nodes with seed_node_id
for i in 1..node_count {
let child = spawn_node(binary, run_dir, i, relay.port, Some(&seed_public_key));
children.push((i, child));
}
SimCluster {
children,
run_dir: run_dir.to_path_buf(),
binary: binary.to_path_buf(),
relay,
seed_public_key,
_node_count: node_count,
}
}
fn kill_node(&mut self, index: usize) {
if let Some(pos) = self.children.iter().position(|(i, _)| *i == index) {
let (_, mut child) = self.children.remove(pos);
let _ = signal_term(child.id());
let _ = child.wait();
}
}
fn restart_node(&mut self, index: usize) {
// Non-seed nodes need the seed's public key; the seed itself doesn't
let seed_id = if index != 0 {
Some(self.seed_public_key.as_str())
} else {
None
};
let child = spawn_node(
&self.binary,
&self.run_dir,
index,
self.relay.port,
seed_id,
);
self.children.push((index, child));
}
}
impl Drop for SimCluster {
fn drop(&mut self) {
for (_, child) in &self.children {
let _ = signal_term(child.id());
}
for (_, child) in &mut self.children {
let _ = child.wait();
}
// relay drops automatically via RelayServer Drop
// NOTE: leaving run_dir for debugging; uncomment below for production
// let _ = fs::remove_dir_all(&self.run_dir);
}
}
fn spawn_node(
binary: &Path,
run_dir: &Path,
index: usize,
relay_port: u16,
seed_public_key: Option<&str>,
) -> Child {
let node_dir = run_dir.join(format!("node-{index}"));
let config_path = write_node_config(&node_dir, index, relay_port, seed_public_key);
let log_path = node_dir.join("stderr.log");
let log_file = fs::File::create(&log_path)
.unwrap_or_else(|e| panic!("failed to create log for node {index}: {e}"));
Command::new(binary)
.args(["--config", &config_path.to_string_lossy()])
.stdout(Stdio::null())
.stderr(Stdio::from(log_file))
.spawn()
.unwrap_or_else(|e| panic!("failed to spawn node {index}: {e}"))
}
fn signal_term(pid: u32) -> std::io::Result<()> {
unsafe { libc::kill(pid as i32, libc::SIGTERM) };
Ok(())
}
// ── HTTP polling (decoupled from distribution crate) ────────────────────
fn poll_distribution(port: u16) -> Option<serde_json::Value> {
let url = format!("http://127.0.0.1:{port}/api/distribution");
let client = reqwest::blocking::Client::builder()
.timeout(Duration::from_secs(2))
.build()
.ok()?;
let resp = client.get(&url).send().ok()?;
if !resp.status().is_success() {
return None;
}
let text = resp.text().ok()?;
if text == "{}" {
return None;
}
serde_json::from_str(&text).ok()
}
fn get_usize(val: &serde_json::Value, key: &str) -> Option<usize> {
val.get(key).and_then(|v| v.as_u64()).map(|n| n as usize)
}
fn wait_for_convergence(
ports: &[u16],
expected_alive: usize,
timeout: Duration,
) -> Result<Duration, String> {
let start = Instant::now();
loop {
if start.elapsed() > timeout {
let mut diag = String::from("Convergence timeout. Last seen: ");
for &port in ports {
match poll_distribution(port) {
Some(snap) => {
let alive = get_usize(&snap, "alive_count").unwrap_or(0);
diag.push_str(&format!("port {port}={alive}, "));
}
None => diag.push_str(&format!("port {port}=unreachable, ")),
}
}
return Err(diag);
}
let all_converged = ports.iter().all(|&port| {
poll_distribution(port)
.and_then(|snap| get_usize(&snap, "alive_count"))
.map(|alive| alive >= expected_alive)
.unwrap_or(false)
});
if all_converged {
return Ok(start.elapsed());
}
thread::sleep(Duration::from_secs(1));
}
}
fn wait_for_death_detection(
ports: &[u16],
max_alive: usize,
timeout: Duration,
) -> Result<Duration, String> {
let start = Instant::now();
loop {
if start.elapsed() > timeout {
let mut diag = String::from("Death detection timeout. Last seen: ");
for &port in ports {
match poll_distribution(port) {
Some(snap) => {
let alive = get_usize(&snap, "alive_count").unwrap_or(0);
diag.push_str(&format!("port {port}={alive} alive, "));
}
None => diag.push_str(&format!("port {port}=unreachable, ")),
}
}
return Err(diag);
}
let all_detected = ports.iter().all(|&port| {
poll_distribution(port)
.and_then(|snap| get_usize(&snap, "alive_count"))
.map(|alive| alive <= max_alive)
.unwrap_or(false)
});
if all_detected {
return Ok(start.elapsed());
}
thread::sleep(Duration::from_secs(1));
}
}
fn wait_for_dead_count(ports: &[u16], min_dead: usize, timeout: Duration) -> bool {
let start = Instant::now();
loop {
if start.elapsed() > timeout {
return false;
}
let any_sees_dead = ports.iter().any(|&port| {
poll_distribution(port)
.and_then(|snap| get_usize(&snap, "dead_count"))
.map(|dead| dead >= min_dead)
.unwrap_or(false)
});
if any_sees_dead {
return true;
}
thread::sleep(Duration::from_secs(1));
}
}
// ── Scenarios ───────────────────────────────────────────────────────────
fn scenario_cluster_convergence(binary: &Path, base_dir: &Path) -> Result<(), String> {
let run_dir = base_dir.join("scenario-1");
let cluster = SimCluster::spawn(binary, &run_dir, NODE_COUNT);
let ports = all_dashboard_ports(NODE_COUNT);
// alive_count excludes self, so each node sees NODE_COUNT - 1 peers
let expected_alive = NODE_COUNT - 1;
let elapsed = wait_for_convergence(&ports, expected_alive, TIMEOUT)?;
println!(" converged in {:.1}s", elapsed.as_secs_f64());
// Verify each node's snapshot
for (i, &port) in ports.iter().enumerate() {
let snap = poll_distribution(port)
.ok_or_else(|| format!("node {i} (port {port}) unreachable after convergence"))?;
let alive = get_usize(&snap, "alive_count").unwrap_or(0);
let routing = get_usize(&snap, "routing_table_size").unwrap_or(0);
if alive < expected_alive {
return Err(format!("node {i} sees {alive} alive, expected >= {expected_alive}"));
}
if routing < NODE_COUNT - 1 {
return Err(format!(
"node {i} has routing_table_size {routing}, expected >= {}",
NODE_COUNT - 1
));
}
}
drop(cluster);
Ok(())
}
fn scenario_node_death_detection(binary: &Path, base_dir: &Path) -> Result<(), String> {
let run_dir = base_dir.join("scenario-2");
let mut cluster = SimCluster::spawn(binary, &run_dir, NODE_COUNT);
let ports = all_dashboard_ports(NODE_COUNT);
let expected_alive = NODE_COUNT - 1;
wait_for_convergence(&ports, expected_alive, TIMEOUT)
.map_err(|e| format!("pre-kill convergence failed: {e}"))?;
// Kill node 2
cluster.kill_node(2);
// Survivors: all except index 2
let survivor_ports: Vec<u16> = (0..NODE_COUNT)
.filter(|&i| i != 2)
.map(dashboard_port)
.collect();
// Wait for alive count to drop (alive_count excludes self, so 5-node cluster
// sees 4 alive; after killing 1, survivors should see <= 3)
let elapsed = wait_for_death_detection(&survivor_ports, NODE_COUNT - 2, TIMEOUT)?;
println!(" detected in {:.1}s", elapsed.as_secs_f64());
// Poll for dead_count — SWIM transitions suspect→dead with a delay
let dead_detected = wait_for_dead_count(&survivor_ports, 1, TIMEOUT);
if !dead_detected {
return Err("no survivor detected a dead member".into());
}
drop(cluster);
Ok(())
}
fn scenario_killed_node_rejoins(binary: &Path, base_dir: &Path) -> Result<(), String> {
let run_dir = base_dir.join("scenario-3");
let mut cluster = SimCluster::spawn(binary, &run_dir, NODE_COUNT);
let ports = all_dashboard_ports(NODE_COUNT);
let expected_alive = NODE_COUNT - 1;
wait_for_convergence(&ports, expected_alive, TIMEOUT)
.map_err(|e| format!("pre-kill convergence failed: {e}"))?;
// Kill node 2
cluster.kill_node(2);
let survivor_ports: Vec<u16> = (0..NODE_COUNT)
.filter(|&i| i != 2)
.map(dashboard_port)
.collect();
wait_for_death_detection(&survivor_ports, NODE_COUNT - 2, TIMEOUT)
.map_err(|e| format!("death detection failed: {e}"))?;
// Restart node 2 — identity persists, so cluster recognizes it
cluster.restart_node(2);
let rejoined_port = dashboard_port(2);
let elapsed = wait_for_convergence(&[rejoined_port], 1, TIMEOUT)?;
println!(" rejoined in {:.1}s", elapsed.as_secs_f64());
let snap = poll_distribution(rejoined_port)
.ok_or("rejoined node unreachable")?;
let alive = get_usize(&snap, "alive_count").unwrap_or(0);
if alive < 1 {
return Err(format!("rejoined node sees {alive} alive, expected >= 1"));
}
drop(cluster);
Ok(())
}
fn scenario_actors_resolvable(binary: &Path, base_dir: &Path) -> Result<(), String> {
let run_dir = base_dir.join("scenario-4");
let cluster = SimCluster::spawn(binary, &run_dir, NODE_COUNT);
let ports = all_dashboard_ports(NODE_COUNT);
let expected_alive = NODE_COUNT - 1;
wait_for_convergence(&ports, expected_alive, TIMEOUT)
.map_err(|e| format!("convergence failed: {e}"))?;
let mut total_directory_entries = 0usize;
for (i, &port) in ports.iter().enumerate() {
let snap = poll_distribution(port)
.ok_or_else(|| format!("node {i} unreachable"))?;
let dir_count = get_usize(&snap, "directory_entry_count").unwrap_or(0);
if dir_count < ACTORS_PER_NODE {
return Err(format!(
"node {i} has {dir_count} directory entries, expected >= {ACTORS_PER_NODE}"
));
}
total_directory_entries += dir_count;
}
let expected_total = NODE_COUNT * ACTORS_PER_NODE;
if total_directory_entries < expected_total {
return Err(format!(
"total directory entries {total_directory_entries}, expected >= {expected_total}"
));
}
println!(" total: {total_directory_entries} directory entries");
drop(cluster);
Ok(())
}
// ── Interactive mode ─────────────────────────────────────────────────────
pub fn run_interactive(node_count: usize) {
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex};
let root = workspace_root();
println!("=== sim-cluster: building swactor (iroh) ===");
let binary = build_swactor(&root);
let base_dir = root.join(".sim-cluster");
let _ = fs::remove_dir_all(&base_dir);
let run_dir = base_dir.join("interactive");
println!("=== sim-cluster: spawning {node_count} nodes ===");
let cluster = SimCluster::spawn(&binary, &run_dir, node_count);
let ports = all_dashboard_ports(node_count);
// Wait for all dashboards
println!("=== sim-cluster: waiting for dashboards ===");
for &port in &ports {
wait_for_dashboard(port, Duration::from_secs(30));
}
// Wait for convergence
println!("=== sim-cluster: waiting for convergence ===");
let expected_alive = node_count - 1;
let convergence = wait_for_convergence(&ports, expected_alive, TIMEOUT);
let converged_msg = match &convergence {
Ok(elapsed) => format!("{node_count} nodes, all converged in {:.1}s", elapsed.as_secs_f64()),
Err(e) => format!("{node_count} nodes, convergence issue: {e}"),
};
// Print summary
println!();
println!("=== sim-cluster ready ===");
println!(" relay: http://127.0.0.1:{}/", cluster.relay.port);
for i in 0..node_count {
let port = dashboard_port(i);
let label = if i == 0 { " (seed)" } else { "" };
println!(" node-{i}: http://127.0.0.1:{port}/{label}");
}
println!(" cluster: {converged_msg}");
println!();
println!(" Logs: .sim-cluster/interactive/node-N/stderr.log");
println!(" Press Ctrl-C to shut down.");
// Block on Ctrl-C
let shutdown = Arc::new((Mutex::new(false), Condvar::new()));
let shutdown2 = Arc::clone(&shutdown);
let flag = Arc::new(AtomicBool::new(false));
let flag2 = Arc::clone(&flag);
unsafe {
let shutdown_ptr = Arc::into_raw(shutdown2) as usize;
let flag_ptr = Arc::into_raw(flag2) as usize;
libc::signal(libc::SIGINT, handler as *const () as libc::sighandler_t);
SHUTDOWN_PTR.store(shutdown_ptr, Ordering::SeqCst);
FLAG_PTR.store(flag_ptr, Ordering::SeqCst);
}
let (lock, cvar) = &*shutdown;
let mut stopped = lock.lock().unwrap();
while !*stopped {
stopped = cvar.wait(stopped).unwrap();
}
println!("\n=== sim-cluster: shutting down ===");
drop(cluster);
println!("=== sim-cluster: stopped ===");
}
// Signal handler support for run_interactive
use std::sync::atomic::{AtomicUsize, Ordering};
static SHUTDOWN_PTR: AtomicUsize = AtomicUsize::new(0);
static FLAG_PTR: AtomicUsize = AtomicUsize::new(0);
extern "C" fn handler(_sig: libc::c_int) {
use std::sync::atomic::AtomicBool;
use std::sync::{Condvar, Mutex};
let flag_ptr = FLAG_PTR.load(Ordering::SeqCst);
if flag_ptr != 0 {
let flag = unsafe { &*(flag_ptr as *const AtomicBool) };
if flag.swap(true, Ordering::SeqCst) {
// Second Ctrl-C — force exit
std::process::exit(1);
}
}
let ptr = SHUTDOWN_PTR.load(Ordering::SeqCst);
if ptr != 0 {
let pair = unsafe { &*(ptr as *const (Mutex<bool>, Condvar)) };
if let Ok(mut stopped) = pair.0.lock() {
*stopped = true;
pair.1.notify_one();
}
}
}
// ── Entry point (test mode) ─────────────────────────────────────────────
pub fn run() {
let root = workspace_root();
let overall_start = Instant::now();
println!("=== sim-cluster: building swactor (iroh) ===");
let binary = build_swactor(&root);
let base_dir = root.join(".sim-cluster");
// Clean any stale runs
let _ = fs::remove_dir_all(&base_dir);
let scenarios: &[(&str, fn(&Path, &Path) -> Result<(), String>)] = &[
("cluster convergence", scenario_cluster_convergence),
("node death detection", scenario_node_death_detection),
("killed node rejoins", scenario_killed_node_rejoins),
("actors resolvable", scenario_actors_resolvable),
];
let total = scenarios.len();
let mut passed = 0usize;
for (i, (name, func)) in scenarios.iter().enumerate() {
println!(
"=== sim-cluster: scenario {}/{total} \u{2014} {name} ===",
i + 1
);
match func(&binary, &base_dir) {
Ok(()) => passed += 1,
Err(e) => {
let elapsed = overall_start.elapsed();
eprintln!(" FAILED: {e}");
eprintln!(
"\n--- FAILED after {:.1}s ({passed}/{total} passed) ---",
elapsed.as_secs_f64()
);
// Leave .sim-cluster for debugging
std::process::exit(1);
}
}
}
// Clean up base dir
let _ = fs::remove_dir_all(&base_dir);
let elapsed = overall_start.elapsed();
println!(
"\n--- All {total} scenario(s) passed in {:.1}s ---",
elapsed.as_secs_f64()
);
}