feat(node): add cross-platform Rust daemon foundation
This commit is contained in:
@@ -0,0 +1,9 @@
|
||||
# Rust workspace build artifacts
|
||||
/target/
|
||||
|
||||
# Repository-local development environment
|
||||
.direnv/
|
||||
.envrc
|
||||
|
||||
# Secrets
|
||||
*.key
|
||||
@@ -0,0 +1,9 @@
|
||||
# Repository implementation rules
|
||||
|
||||
## Cross-platform support
|
||||
|
||||
- Treat cross-platform behavior as a design requirement for daemon code. Keep domain logic independent of operating-system-specific paths and APIs.
|
||||
- Never hardcode home-directory paths, path separators, or assumptions about a writable current directory. Use `Path`/`PathBuf` and the shared platform helpers in `apps/sisyphusd/src/platform.rs`.
|
||||
- Put persistent node data in the OS-appropriate application data directory. The SQLite database contains the private node identity, so protect the database and its containing directory with restrictive permissions where the platform supports them; rely on the user's inherited ACL on Windows unless a dedicated ACL implementation is introduced.
|
||||
- Isolate unavoidable OS-specific behavior behind small helpers and `#[cfg(...)]` blocks. Keep a portable fallback where possible.
|
||||
- Add tests for path handling and platform-specific behavior. Verify supported targets in CI before claiming support; the Nix dev shell currently targets Linux and macOS.
|
||||
Generated
+5000
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,8 @@
|
||||
[workspace]
|
||||
members = ["apps/sisyphusd"]
|
||||
resolver = "2"
|
||||
|
||||
[workspace.package]
|
||||
edition = "2024"
|
||||
license = "MIT"
|
||||
version = "0.1.0"
|
||||
@@ -0,0 +1,6 @@
|
||||
# Local daemon state may contain the node's private identity.
|
||||
/target/
|
||||
/data/
|
||||
*.sqlite3
|
||||
*.sqlite3-shm
|
||||
*.sqlite3-wal
|
||||
@@ -1 +0,0 @@
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
[package]
|
||||
name = "sisyphusd"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
|
||||
[dependencies]
|
||||
async-trait = "0.1"
|
||||
anyhow = "1"
|
||||
clap = { version = "4", features = ["derive"] }
|
||||
directories = "6"
|
||||
futures-util = "0.3"
|
||||
libp2p = { version = "0.57", features = [
|
||||
"ed25519",
|
||||
"identify",
|
||||
"kad",
|
||||
"macros",
|
||||
"mdns",
|
||||
"noise",
|
||||
"ping",
|
||||
"quic",
|
||||
"tcp",
|
||||
"tokio",
|
||||
"yamux",
|
||||
] }
|
||||
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
prost = "0.14.1"
|
||||
sqlx = { version = "0.8.6", default-features = false, features = [
|
||||
"macros",
|
||||
"migrate",
|
||||
"runtime-tokio",
|
||||
"sqlite",
|
||||
] }
|
||||
tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal"] }
|
||||
tokio-stream = { version = "0.1", features = ["net"] }
|
||||
tonic = { version = "0.14.6", features = ["transport"] }
|
||||
tonic-prost = "0.14.6"
|
||||
tracing = "0.1"
|
||||
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
|
||||
|
||||
[dev-dependencies]
|
||||
tempfile = "3"
|
||||
|
||||
[build-dependencies]
|
||||
tonic-prost-build = "0.14.6"
|
||||
@@ -0,0 +1,110 @@
|
||||
# sisyphusd
|
||||
|
||||
`sisyphusd` is the Sisyphus node daemon. It runs headless or underneath a desktop client and owns node identity, P2P networking, local state, and—over time—hardware discovery, scheduling, workload execution, and AI orchestration.
|
||||
|
||||
This document covers daemon-specific architecture and development. The product vision, network economy, and V0 scope are described in the [repository README](../../README.md).
|
||||
|
||||
## Architecture and boundaries
|
||||
|
||||
One daemon binary can run different node roles. Coordinator and worker are capabilities inside `sisyphusd`, not separate services or binaries. A coordinator may also execute work locally; a headless daemon can contribute as a worker.
|
||||
|
||||
```text
|
||||
Clients (Electron / Web / agents)
|
||||
│
|
||||
daemon client protocol (planned)
|
||||
│
|
||||
sisyphusd
|
||||
┌──────────┼──────────┐
|
||||
│ │ │
|
||||
storage network compute (planned)
|
||||
```
|
||||
|
||||
Keep domain logic independent of the transport, persistence implementation, and operating system. Add interfaces at boundaries where they enable real alternatives or testing; avoid speculative layers for components that do not exist yet.
|
||||
|
||||
## Local gRPC API
|
||||
|
||||
The daemon exposes a small Protobuf/gRPC API for local clients. Definitions live in the repository's [`proto/sisyphus/node/v1/node.proto`](../../proto/sisyphus/node/v1/node.proto); Rust client/server bindings are generated at build time using `protoc` from the Nix development shell.
|
||||
|
||||
V0 exposes:
|
||||
|
||||
- `NodeService.GetNodeInfo`: peer ID, daemon version, and listen addresses.
|
||||
- `NodeService.ListPeers`: connected peers plus saved bootstrap peers, their connection state, and known addresses.
|
||||
- `NodeService.WatchPeers`: an initial peer snapshot followed by a new full snapshot only when peer state changes; each update carries a monotonically increasing revision.
|
||||
- `NodeService.GetBootstrapPeers` / `SetBootstrapPeers`: read or atomically replace the node's saved bootstrap address book. Addresses must include a matching terminal `/p2p/<peer-id>`.
|
||||
|
||||
The server binds to `127.0.0.1:50051` by default. `--api-listen` can select another loopback address/port; non-loopback binding is rejected until authentication is implemented. This is a local desktop/daemon API, not a public or browser-ready gRPC-Web endpoint. A hosted Web UI will need an authenticated gateway/relay path; this API alone does not provide remote access.
|
||||
|
||||
`GetNodeInfo.country_code` is an optional, approximate country inferred at daemon startup from the node's public egress IP using ipapi.co over HTTPS. Only the two-letter country code is returned to local clients; the daemon does not persist or return the IP. The lookup is best-effort and has a three-second timeout, so GeoIP outages do not prevent node startup. The provider necessarily sees the source IP of the HTTPS request; this prototype integration is for development and must be replaced with a production-appropriate provider or local GeoIP database before broad deployment. VPNs, relays, and some network setups can make the estimate inaccurate.
|
||||
|
||||
Bootstrap settings are stored in the node's SQLite database and updates are applied to the running networking loop. Removing a peer from the bootstrap list prevents future bootstrap dials but does not forcibly close an existing connection. Peer status is streamed; chat/job streaming, general node settings, jobs, and P2P application messages are not part of this first contract. Watch updates are snapshots rather than deltas, so a client that reconnects can resume from a fresh initial state. Protobuf fields use stable field numbers; never reuse a field number after removing a field.
|
||||
|
||||
The gRPC listener is bound before the daemon reports startup, so address conflicts fail startup with a contextual error. On Ctrl-C it stops accepting requests, signals active peer streams to finish, then waits for graceful server shutdown.
|
||||
|
||||
## P2P networking
|
||||
|
||||
Daemon-to-daemon connectivity uses Rust `libp2p`. The current scaffold configures TCP and QUIC transports, Noise encryption, Yamux stream multiplexing, Identify and Ping, mDNS local discovery, and Kademlia primitives. The local client API uses Protocol Buffers/gRPC; the separate application job protocol over P2P is not implemented yet.
|
||||
|
||||
The scaffold supports encrypted peer connections and basic discovery. Workload transfer, relay support, robust NAT traversal, bootstrap-node operations, and the compute/job protocol are future work. Kademlia currently uses an in-memory address store, so learned DHT state does not survive restarts. The local gRPC client API is separate from this P2P transport; it does not tunnel gRPC between nodes.
|
||||
|
||||
## Local state and storage
|
||||
|
||||
Each daemon owns a separate local SQLite database. Nodes do not share a database. Storage access is behind the `StateStore` boundary so daemon domains do not depend directly on SQLx or SQLite.
|
||||
|
||||
Current implementation:
|
||||
|
||||
- SQLx SQLite with embedded, versioned migrations.
|
||||
- WAL journal mode and a bounded connection pool for this single-node process.
|
||||
- Persistent libp2p identity stored as a singleton private-key record. Concurrent first startup converges on the same stored identity.
|
||||
- Database and data directory are restricted on Unix; Windows relies on inherited user ACLs.
|
||||
|
||||
SQLite is for local node state, not a distributed database or consensus ledger. Keep large datasets, model artifacts, and job outputs in files/blob storage and persist identifiers, hashes, and metadata in SQLite. If a blockchain is introduced, its consensus-critical state and storage engine should be designed with that protocol rather than treating the local node DB as authoritative network state.
|
||||
|
||||
### Data location and identity
|
||||
|
||||
By default the daemon stores `node.sqlite3` under the operating system's local application-data directory (`sisyphus/`). `--data-dir` selects a different directory. Reusing the same database preserves the node's `PeerId`; deleting it causes a new identity to be generated.
|
||||
|
||||
Bootstrap peers supplied with `--bootstrap` are added to the node's persistent bootstrap list and dialed on subsequent starts. The CLI option is additive. Use `--clear-bootstrap-peers` to remove the saved list; combine it with one or more `--bootstrap` options to replace the list. `--list-bootstrap-peers` prints the persisted address book and exits. While running, the daemon logs peer connection and disconnection events.
|
||||
|
||||
The database contains the node's private identity key and is not encrypted by the application. Protect the database and backups as secrets. Do not copy one node database to multiple live nodes: that duplicates identity.
|
||||
|
||||
## AI provider boundary
|
||||
|
||||
AI provider configuration belongs to the node, not the UI or remote workers. V0 is intended to support bring-your-own-key with initial adapters for Ollama and the OpenAI API. Provider calls and credentials stay on the configured node; the UI configures it and presents responses. Keep provider integrations behind a small replaceable interface that can support streaming, tool calls, and structured plans.
|
||||
|
||||
## Cross-platform rules
|
||||
|
||||
- Treat cross-platform behavior as a design requirement. Keep domain logic independent of OS paths and APIs.
|
||||
- Never hardcode home-directory paths, path separators, or assumptions about a writable current directory. Use `Path`/`PathBuf` and shared helpers in `src/platform.rs`.
|
||||
- Store persistent node data in the OS-appropriate application-data directory. Protect the database and its containing directory with restrictive permissions where supported; rely on the user's inherited ACL on Windows unless a dedicated ACL implementation is introduced.
|
||||
- Isolate unavoidable OS-specific behavior behind small helpers and `#[cfg(...)]` blocks. Keep a portable fallback where possible.
|
||||
- Add tests for path handling and platform-specific behavior. Verify supported targets in CI before claiming support; the Nix dev shell currently targets Linux and macOS.
|
||||
|
||||
## Development
|
||||
|
||||
The repository Nix flake provides the Rust toolchain, formatter, linter, language server, Protobuf compiler, and native build tools.
|
||||
|
||||
```sh
|
||||
nix develop
|
||||
cargo run -p sisyphusd
|
||||
```
|
||||
|
||||
Run checks from the repository root:
|
||||
|
||||
```sh
|
||||
cargo fmt --check
|
||||
cargo check -p sisyphusd
|
||||
cargo test -p sisyphusd
|
||||
```
|
||||
|
||||
## Running and connecting nodes
|
||||
|
||||
On first startup, the daemon initializes the SQLite database and node identity. It logs its `PeerId`, listen multiaddresses, bootstrap peers, and live peer connection/disconnection events. To run a second local node and connect it to the first, use the first node's full `/p2p/<peer-id>` address:
|
||||
|
||||
```sh
|
||||
cargo run -p sisyphusd -- \
|
||||
--data-dir ./sisyphus-peer-b \
|
||||
--listen /ip4/0.0.0.0/tcp/7410 \
|
||||
--bootstrap /ip4/127.0.0.1/tcp/7400/p2p/<peer-id>
|
||||
```
|
||||
|
||||
Use `cargo run -p sisyphusd -- --help` to see current command-line options. For example, `cargo run -p sisyphusd -- --list-bootstrap-peers` inspects the saved peer address book.
|
||||
@@ -1 +0,0 @@
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
fn main() {
|
||||
println!("cargo:rerun-if-changed=migrations");
|
||||
println!("cargo:rerun-if-changed=../../proto/sisyphus/node/v1/node.proto");
|
||||
|
||||
tonic_prost_build::configure()
|
||||
.compile_protos(
|
||||
&["../../proto/sisyphus/node/v1/node.proto"],
|
||||
&["../../proto"],
|
||||
)
|
||||
.expect("failed to compile daemon protobuf definitions");
|
||||
}
|
||||
@@ -1 +0,0 @@
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
CREATE TABLE node_identity (
|
||||
singleton_id INTEGER PRIMARY KEY NOT NULL CHECK (singleton_id = 1),
|
||||
encoding_version INTEGER NOT NULL,
|
||||
private_key BLOB NOT NULL,
|
||||
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
@@ -0,0 +1,6 @@
|
||||
CREATE TABLE bootstrap_peers (
|
||||
peer_id TEXT NOT NULL,
|
||||
address TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
PRIMARY KEY (peer_id, address)
|
||||
);
|
||||
@@ -0,0 +1,4 @@
|
||||
CREATE TABLE trusted_compute_peers (
|
||||
peer_id TEXT PRIMARY KEY,
|
||||
trusted_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
@@ -1 +0,0 @@
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
use std::path::PathBuf;
|
||||
|
||||
use clap::Parser;
|
||||
|
||||
#[derive(Debug, Parser)]
|
||||
#[command(name = "sisyphusd", about = "Sisyphus peer-to-peer node")]
|
||||
pub(crate) struct Args {
|
||||
/// Multiaddresses to listen on. Can be supplied more than once.
|
||||
#[arg(
|
||||
long = "listen",
|
||||
default_values = [
|
||||
"/ip4/0.0.0.0/tcp/7400",
|
||||
"/ip4/0.0.0.0/udp/7401/quic-v1"
|
||||
]
|
||||
)]
|
||||
pub(crate) listen: Vec<String>,
|
||||
|
||||
/// Bootstrap peer multiaddress, including its /p2p/<peer-id>. Can be supplied more than once.
|
||||
#[arg(long = "bootstrap")]
|
||||
pub(crate) bootstrap_peers: Vec<String>,
|
||||
|
||||
/// Remove all saved bootstrap peers before adding any supplied with --bootstrap.
|
||||
#[arg(long)]
|
||||
pub(crate) clear_bootstrap_peers: bool,
|
||||
|
||||
/// Print the saved bootstrap peer address book and exit.
|
||||
#[arg(long)]
|
||||
pub(crate) list_bootstrap_peers: bool,
|
||||
|
||||
/// Directory for this node's local SQLite database (defaults to the OS application data directory).
|
||||
#[arg(long)]
|
||||
pub(crate) data_dir: Option<PathBuf>,
|
||||
|
||||
/// Loopback address for the local gRPC API (remote binding is disabled until authentication exists).
|
||||
#[arg(long, default_value = "127.0.0.1:50051")]
|
||||
pub(crate) api_listen: String,
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
use std::time::Duration;
|
||||
|
||||
use serde::Deserialize;
|
||||
use tracing::debug;
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct GeoIpResponse {
|
||||
country_code: Option<String>,
|
||||
}
|
||||
|
||||
/// Detect the country of the node's public egress IP without returning or storing that IP.
|
||||
/// GeoIP is best-effort and must never prevent the daemon from starting.
|
||||
pub(crate) async fn detect_country_code() -> Option<String> {
|
||||
let client = match reqwest::Client::builder()
|
||||
.timeout(Duration::from_secs(3))
|
||||
.user_agent(concat!("sisyphusd/", env!("CARGO_PKG_VERSION")))
|
||||
.build()
|
||||
{
|
||||
Ok(client) => client,
|
||||
Err(error) => {
|
||||
debug!(%error, "could not initialize GeoIP client");
|
||||
return None;
|
||||
}
|
||||
};
|
||||
|
||||
let result = async {
|
||||
let response = client
|
||||
.get("https://ipapi.co/json/")
|
||||
.send()
|
||||
.await?
|
||||
.error_for_status()?;
|
||||
let location: GeoIpResponse = response.json().await?;
|
||||
Ok::<_, reqwest::Error>(location.country_code)
|
||||
}
|
||||
.await;
|
||||
|
||||
match result {
|
||||
Ok(Some(country_code)) if is_iso_alpha2(&country_code) => {
|
||||
Some(country_code.to_ascii_uppercase())
|
||||
}
|
||||
Ok(_) => None,
|
||||
Err(error) => {
|
||||
debug!(%error, "GeoIP lookup unavailable; starting without a country code");
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn is_iso_alpha2(value: &str) -> bool {
|
||||
value.len() == 2 && value.bytes().all(|byte| byte.is_ascii_alphabetic())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::is_iso_alpha2;
|
||||
|
||||
#[test]
|
||||
fn accepts_only_two_ascii_letters() {
|
||||
assert!(is_iso_alpha2("US"));
|
||||
assert!(is_iso_alpha2("il"));
|
||||
assert!(!is_iso_alpha2("USA"));
|
||||
assert!(!is_iso_alpha2("1L"));
|
||||
assert!(!is_iso_alpha2("ישראל"));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
mod cli;
|
||||
mod geolocation;
|
||||
mod networking;
|
||||
mod observability;
|
||||
mod platform;
|
||||
mod rpc;
|
||||
mod storage;
|
||||
|
||||
use clap::Parser;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
observability::init();
|
||||
networking::run(cli::Args::parse()).await
|
||||
}
|
||||
@@ -0,0 +1,250 @@
|
||||
mod swarm;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use futures_util::StreamExt;
|
||||
use libp2p::{Multiaddr, PeerId, Swarm};
|
||||
use std::{collections::BTreeMap, net::SocketAddr};
|
||||
use tokio::sync::{mpsc, oneshot, watch};
|
||||
use tracing::{info, warn};
|
||||
|
||||
use crate::{
|
||||
cli::Args,
|
||||
rpc::{self, NodeSnapshot, PeerSnapshot},
|
||||
storage::BootstrapPeer,
|
||||
};
|
||||
|
||||
pub(crate) async fn run(args: Args) -> Result<()> {
|
||||
let data_dir = args.data_dir;
|
||||
let storage = crate::storage::open(data_dir).await?;
|
||||
|
||||
if args.list_bootstrap_peers {
|
||||
for peer in storage.load_bootstrap_peers().await? {
|
||||
println!("{} {}", peer.peer_id, peer.address);
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let keypair = storage.load_or_create_identity().await?;
|
||||
let local_peer_id = keypair.public().to_peer_id();
|
||||
let mut swarm = swarm::build_swarm(keypair)?;
|
||||
let country_code = crate::geolocation::detect_country_code().await;
|
||||
if let Some(country_code) = &country_code {
|
||||
info!(%country_code, "detected approximate country from public egress IP");
|
||||
}
|
||||
let mut trusted_peers = storage
|
||||
.load_compute_trusted_peers()
|
||||
.await?
|
||||
.into_iter()
|
||||
.collect();
|
||||
|
||||
let api_address: SocketAddr = args
|
||||
.api_listen
|
||||
.parse()
|
||||
.with_context(|| format!("invalid gRPC API listen address: {}", args.api_listen))?;
|
||||
if !api_address.ip().is_loopback() {
|
||||
anyhow::bail!(
|
||||
"gRPC API can only bind to a loopback address until authentication is implemented"
|
||||
);
|
||||
}
|
||||
|
||||
if args.clear_bootstrap_peers {
|
||||
storage.clear_bootstrap_peers().await?;
|
||||
info!("cleared saved bootstrap peers");
|
||||
}
|
||||
|
||||
for listen_address in args.listen {
|
||||
let address: Multiaddr = listen_address
|
||||
.parse()
|
||||
.with_context(|| format!("invalid listen multiaddress: {listen_address}"))?;
|
||||
swarm
|
||||
.listen_on(address.clone())
|
||||
.with_context(|| format!("failed to listen on {address}"))?;
|
||||
}
|
||||
|
||||
for bootstrap_address in args.bootstrap_peers {
|
||||
let address: Multiaddr = bootstrap_address
|
||||
.parse()
|
||||
.with_context(|| format!("invalid bootstrap multiaddress: {bootstrap_address}"))?;
|
||||
let peer_id = swarm::peer_id_from_address(&address)
|
||||
.with_context(|| format!("bootstrap address must include /p2p/<peer-id>: {address}"))?;
|
||||
|
||||
storage
|
||||
.save_bootstrap_peer(&BootstrapPeer { peer_id, address })
|
||||
.await?;
|
||||
}
|
||||
|
||||
let bootstrap_peers = storage.load_bootstrap_peers().await?;
|
||||
let (bootstrap_sender, mut bootstrap_updates) = watch::channel(bootstrap_peers.clone());
|
||||
let mut known_peers = BTreeMap::new();
|
||||
for peer in &bootstrap_peers {
|
||||
let peer_id = peer.peer_id;
|
||||
let address = &peer.address;
|
||||
info!(%peer_id, %address, "dialing saved bootstrap peer");
|
||||
known_peers
|
||||
.entry(peer_id)
|
||||
.or_insert_with(Vec::new)
|
||||
.push(address.to_string());
|
||||
swarm::add_peer_address(&mut swarm, peer_id, address.clone());
|
||||
swarm
|
||||
.dial(address.clone())
|
||||
.with_context(|| format!("failed to dial saved bootstrap peer {address}"))?;
|
||||
}
|
||||
|
||||
let (state_sender, state_receiver) = watch::channel(NodeSnapshot {
|
||||
peer_id: local_peer_id.to_string(),
|
||||
daemon_version: env!("CARGO_PKG_VERSION").to_owned(),
|
||||
country_code: country_code.clone(),
|
||||
..NodeSnapshot::default()
|
||||
});
|
||||
let (connect_sender, mut connect_receiver) = mpsc::channel::<rpc::ConnectCommand>(32);
|
||||
let (trust_sender, mut trust_receiver) = mpsc::channel::<(PeerId, bool)>(32);
|
||||
publish_snapshot(
|
||||
&swarm,
|
||||
local_peer_id,
|
||||
&known_peers,
|
||||
&trusted_peers,
|
||||
&state_sender,
|
||||
);
|
||||
let rpc_server = rpc::RpcServer::bind(
|
||||
api_address,
|
||||
state_receiver,
|
||||
storage,
|
||||
bootstrap_sender,
|
||||
connect_sender,
|
||||
trust_sender,
|
||||
)
|
||||
.await?;
|
||||
info!(%local_peer_id, "Sisyphus node started");
|
||||
let (rpc_shutdown, rpc_shutdown_receiver) = oneshot::channel();
|
||||
let rpc_server = rpc_server.serve(rpc_shutdown_receiver);
|
||||
tokio::pin!(rpc_server);
|
||||
|
||||
let shutdown = tokio::signal::ctrl_c();
|
||||
tokio::pin!(shutdown);
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
event = swarm.select_next_some() => {
|
||||
swarm::handle_swarm_event(event, &mut swarm, local_peer_id);
|
||||
publish_snapshot(&swarm, local_peer_id, &known_peers, &trusted_peers, &state_sender);
|
||||
},
|
||||
command = connect_receiver.recv() => {
|
||||
if let Some(command) = command {
|
||||
let (peer_id, address, reply) = command;
|
||||
swarm::add_peer_address(&mut swarm, peer_id, address.clone());
|
||||
let result = swarm.dial(address.clone()).map_err(|error| error.to_string());
|
||||
if result.is_ok() {
|
||||
known_peers.entry(peer_id).or_insert_with(Vec::new).push(address.to_string());
|
||||
}
|
||||
let _ = reply.send(result);
|
||||
publish_snapshot(&swarm, local_peer_id, &known_peers, &trusted_peers, &state_sender);
|
||||
}
|
||||
},
|
||||
update = trust_receiver.recv() => {
|
||||
if let Some((peer_id, trusted)) = update {
|
||||
if trusted { trusted_peers.insert(peer_id); } else { trusted_peers.remove(&peer_id); }
|
||||
publish_snapshot(&swarm, local_peer_id, &known_peers, &trusted_peers, &state_sender);
|
||||
}
|
||||
},
|
||||
result = bootstrap_updates.changed() => {
|
||||
result.context("bootstrap peer update channel closed")?;
|
||||
let peers = bootstrap_updates.borrow_and_update().clone();
|
||||
known_peers = address_book(&peers);
|
||||
for peer in peers {
|
||||
if !swarm.is_connected(&peer.peer_id) {
|
||||
swarm::add_peer_address(&mut swarm, peer.peer_id, peer.address.clone());
|
||||
if let Err(error) = swarm.dial(peer.address.clone()) {
|
||||
warn!(peer_id = %peer.peer_id, address = %peer.address, %error, "could not dial updated bootstrap peer");
|
||||
}
|
||||
}
|
||||
}
|
||||
publish_snapshot(&swarm, local_peer_id, &known_peers, &trusted_peers, &state_sender);
|
||||
},
|
||||
result = &mut rpc_server => {
|
||||
result?;
|
||||
break;
|
||||
}
|
||||
result = &mut shutdown => {
|
||||
result.context("failed to listen for shutdown signal")?;
|
||||
info!("shutdown requested");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let _ = rpc_shutdown.send(());
|
||||
drop(state_sender);
|
||||
rpc_server.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn address_book(peers: &[BootstrapPeer]) -> BTreeMap<PeerId, Vec<String>> {
|
||||
let mut address_book = BTreeMap::new();
|
||||
for peer in peers {
|
||||
address_book
|
||||
.entry(peer.peer_id)
|
||||
.or_insert_with(Vec::new)
|
||||
.push(peer.address.to_string());
|
||||
}
|
||||
address_book
|
||||
}
|
||||
|
||||
fn publish_snapshot(
|
||||
swarm: &Swarm<swarm::SisyphusBehaviour>,
|
||||
local_peer_id: PeerId,
|
||||
known_peers: &BTreeMap<PeerId, Vec<String>>,
|
||||
trusted_peers: &std::collections::BTreeSet<PeerId>,
|
||||
state_sender: &watch::Sender<NodeSnapshot>,
|
||||
) {
|
||||
let mut peers: BTreeMap<PeerId, PeerSnapshot> = known_peers
|
||||
.iter()
|
||||
.map(|(peer_id, addresses)| {
|
||||
(
|
||||
*peer_id,
|
||||
PeerSnapshot {
|
||||
peer_id: peer_id.to_string(),
|
||||
connected: false,
|
||||
known_addresses: addresses.clone(),
|
||||
trusted_for_compute: trusted_peers.contains(peer_id),
|
||||
},
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
|
||||
for peer_id in swarm.connected_peers() {
|
||||
peers
|
||||
.entry(*peer_id)
|
||||
.and_modify(|peer| peer.connected = true)
|
||||
.or_insert_with(|| PeerSnapshot {
|
||||
peer_id: peer_id.to_string(),
|
||||
connected: true,
|
||||
known_addresses: Vec::new(),
|
||||
trusted_for_compute: trusted_peers.contains(peer_id),
|
||||
});
|
||||
}
|
||||
|
||||
let peers: Vec<_> = peers.into_values().collect();
|
||||
let current = state_sender.borrow().clone();
|
||||
let peer_revision = if peers == current.peers {
|
||||
current.peer_revision
|
||||
} else {
|
||||
current.peer_revision.saturating_add(1)
|
||||
};
|
||||
|
||||
state_sender.send_replace(NodeSnapshot {
|
||||
peer_id: local_peer_id.to_string(),
|
||||
daemon_version: env!("CARGO_PKG_VERSION").to_owned(),
|
||||
country_code: current.country_code.clone(),
|
||||
listen_addresses: swarm
|
||||
.listeners()
|
||||
.map(|address| {
|
||||
address
|
||||
.clone()
|
||||
.with(libp2p::multiaddr::Protocol::P2p(local_peer_id))
|
||||
.to_string()
|
||||
})
|
||||
.collect(),
|
||||
peers,
|
||||
peer_revision,
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,152 @@
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use libp2p::{
|
||||
Multiaddr, PeerId, Swarm, SwarmBuilder, identify, identity,
|
||||
kad::{self, store::MemoryStore},
|
||||
mdns, noise, ping,
|
||||
swarm::{NetworkBehaviour, SwarmEvent},
|
||||
tcp, yamux,
|
||||
};
|
||||
use tracing::{debug, info, warn};
|
||||
|
||||
const IDENTIFY_PROTOCOL: &str = "/sisyphus/node/1.0.0";
|
||||
|
||||
#[derive(NetworkBehaviour)]
|
||||
pub(super) struct SisyphusBehaviour {
|
||||
identify: identify::Behaviour,
|
||||
ping: ping::Behaviour,
|
||||
mdns: mdns::tokio::Behaviour,
|
||||
kademlia: kad::Behaviour<MemoryStore>,
|
||||
}
|
||||
|
||||
pub(super) fn build_swarm(keypair: identity::Keypair) -> Result<Swarm<SisyphusBehaviour>> {
|
||||
let swarm = SwarmBuilder::with_existing_identity(keypair)
|
||||
.with_tokio()
|
||||
.with_tcp(
|
||||
tcp::Config::default(),
|
||||
noise::Config::new,
|
||||
yamux::Config::default,
|
||||
)
|
||||
.context("failed to configure TCP transport")?
|
||||
.with_quic()
|
||||
.with_behaviour(|key| {
|
||||
let peer_id = key.public().to_peer_id();
|
||||
let identify = identify::Behaviour::new(identify::Config::new(
|
||||
IDENTIFY_PROTOCOL.to_owned(),
|
||||
key.public(),
|
||||
));
|
||||
let ping =
|
||||
ping::Behaviour::new(ping::Config::new().with_interval(Duration::from_secs(15)));
|
||||
let mdns = mdns::tokio::Behaviour::new(mdns::Config::default(), peer_id)
|
||||
.expect("failed to initialize mDNS discovery");
|
||||
let mut kademlia = kad::Behaviour::new(peer_id, MemoryStore::new(peer_id));
|
||||
kademlia.set_mode(Some(kad::Mode::Server));
|
||||
|
||||
SisyphusBehaviour {
|
||||
identify,
|
||||
ping,
|
||||
mdns,
|
||||
kademlia,
|
||||
}
|
||||
})
|
||||
.context("failed to configure libp2p behaviours")?
|
||||
.build();
|
||||
|
||||
Ok(swarm)
|
||||
}
|
||||
|
||||
pub(super) fn handle_swarm_event(
|
||||
event: SwarmEvent<SisyphusBehaviourEvent>,
|
||||
swarm: &mut Swarm<SisyphusBehaviour>,
|
||||
local_peer_id: PeerId,
|
||||
) {
|
||||
match event {
|
||||
SwarmEvent::NewListenAddr { address, .. } => {
|
||||
info!(address = %address.with(libp2p::multiaddr::Protocol::P2p(local_peer_id)), "listening");
|
||||
}
|
||||
SwarmEvent::ConnectionEstablished {
|
||||
peer_id, endpoint, ..
|
||||
} => {
|
||||
info!(%peer_id, ?endpoint, "peer connection established");
|
||||
}
|
||||
SwarmEvent::ConnectionClosed { peer_id, cause, .. } => {
|
||||
info!(%peer_id, ?cause, "peer connection closed");
|
||||
}
|
||||
SwarmEvent::Behaviour(SisyphusBehaviourEvent::Mdns(mdns::Event::Discovered(peers))) => {
|
||||
for (peer_id, address) in peers {
|
||||
info!(%peer_id, %address, "discovered peer on local network");
|
||||
let address = with_peer_id(address, peer_id);
|
||||
swarm
|
||||
.behaviour_mut()
|
||||
.kademlia
|
||||
.add_address(&peer_id, address.clone());
|
||||
if let Err(error) = swarm.dial(address.clone()) {
|
||||
debug!(%peer_id, %address, %error, "could not dial discovered peer");
|
||||
}
|
||||
}
|
||||
}
|
||||
SwarmEvent::Behaviour(SisyphusBehaviourEvent::Mdns(mdns::Event::Expired(peers))) => {
|
||||
for (peer_id, address) in peers {
|
||||
debug!(%peer_id, %address, "local-network peer advertisement expired");
|
||||
}
|
||||
}
|
||||
SwarmEvent::Behaviour(SisyphusBehaviourEvent::Identify(identify::Event::Received {
|
||||
peer_id,
|
||||
info,
|
||||
..
|
||||
})) => {
|
||||
info!(%peer_id, agent = %info.agent_version, protocol = %info.protocol_version, "identified peer");
|
||||
for address in info.listen_addrs {
|
||||
swarm
|
||||
.behaviour_mut()
|
||||
.kademlia
|
||||
.add_address(&peer_id, with_peer_id(address, peer_id));
|
||||
}
|
||||
if let Err(error) = swarm.behaviour_mut().kademlia.bootstrap() {
|
||||
debug!(%peer_id, %error, "DHT bootstrap is waiting for known peers");
|
||||
}
|
||||
}
|
||||
SwarmEvent::Behaviour(SisyphusBehaviourEvent::Ping(ping::Event {
|
||||
peer, result, ..
|
||||
})) => match result {
|
||||
Ok(rtt) => debug!(%peer, ?rtt, "peer ping succeeded"),
|
||||
Err(error) => warn!(%peer, %error, "peer ping failed"),
|
||||
},
|
||||
SwarmEvent::Behaviour(SisyphusBehaviourEvent::Kademlia(event)) => {
|
||||
debug!(?event, "DHT event");
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn peer_id_from_address(address: &Multiaddr) -> Option<PeerId> {
|
||||
address.iter().find_map(|protocol| match protocol {
|
||||
libp2p::multiaddr::Protocol::P2p(peer_id) => Some(peer_id),
|
||||
_ => None,
|
||||
})
|
||||
}
|
||||
|
||||
pub(super) fn add_peer_address(
|
||||
swarm: &mut Swarm<SisyphusBehaviour>,
|
||||
peer_id: PeerId,
|
||||
address: Multiaddr,
|
||||
) {
|
||||
swarm
|
||||
.behaviour_mut()
|
||||
.kademlia
|
||||
.add_address(&peer_id, address);
|
||||
}
|
||||
|
||||
fn with_peer_id(address: Multiaddr, peer_id: PeerId) -> Multiaddr {
|
||||
let already_qualified = matches!(
|
||||
address.iter().last(),
|
||||
Some(libp2p::multiaddr::Protocol::P2p(address_peer_id)) if address_peer_id == peer_id
|
||||
);
|
||||
|
||||
if already_qualified {
|
||||
address
|
||||
} else {
|
||||
address.with(libp2p::multiaddr::Protocol::P2p(peer_id))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
pub(crate) fn init() {
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(
|
||||
tracing_subscriber::EnvFilter::try_from_default_env()
|
||||
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
|
||||
)
|
||||
.init();
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
use std::path::PathBuf;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use directories::BaseDirs;
|
||||
|
||||
const APP_DIRECTORY: &str = "sisyphus";
|
||||
const DATABASE_FILE: &str = "node.sqlite3";
|
||||
|
||||
/// Return the platform-appropriate local data directory for this node.
|
||||
pub(crate) fn app_data_dir() -> Result<PathBuf> {
|
||||
let base_dirs = BaseDirs::new().context("could not locate the user's data directory")?;
|
||||
Ok(base_dirs.data_local_dir().join(APP_DIRECTORY))
|
||||
}
|
||||
|
||||
pub(crate) fn database_path(data_dir: Option<PathBuf>) -> Result<PathBuf> {
|
||||
let data_dir = match data_dir {
|
||||
Some(path) => path,
|
||||
None => app_data_dir()?,
|
||||
};
|
||||
|
||||
Ok(data_dir.join(DATABASE_FILE))
|
||||
}
|
||||
@@ -0,0 +1,655 @@
|
||||
use std::net::SocketAddr;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use futures_util::{Stream, stream};
|
||||
use libp2p::{Multiaddr, PeerId, multiaddr::Protocol};
|
||||
use tokio::{
|
||||
net::TcpListener,
|
||||
sync::{mpsc, oneshot, watch},
|
||||
};
|
||||
use tokio_stream::wrappers::TcpListenerStream;
|
||||
use tonic::{Request, Response, Status, transport::Server};
|
||||
use tracing::info;
|
||||
|
||||
use crate::storage::{BootstrapPeer, StateStore};
|
||||
|
||||
pub(crate) type ConnectCommand = (PeerId, Multiaddr, oneshot::Sender<Result<(), String>>);
|
||||
|
||||
pub(crate) mod proto {
|
||||
tonic::include_proto!("sisyphus.node.v1");
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default)]
|
||||
pub(crate) struct NodeSnapshot {
|
||||
pub(crate) peer_id: String,
|
||||
pub(crate) daemon_version: String,
|
||||
pub(crate) country_code: Option<String>,
|
||||
pub(crate) listen_addresses: Vec<String>,
|
||||
pub(crate) peers: Vec<PeerSnapshot>,
|
||||
pub(crate) peer_revision: u64,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, PartialEq, Eq)]
|
||||
pub(crate) struct PeerSnapshot {
|
||||
pub(crate) peer_id: String,
|
||||
pub(crate) connected: bool,
|
||||
pub(crate) known_addresses: Vec<String>,
|
||||
pub(crate) trusted_for_compute: bool,
|
||||
}
|
||||
|
||||
pub(crate) struct RpcServer {
|
||||
address: SocketAddr,
|
||||
incoming: TcpListenerStream,
|
||||
state: watch::Receiver<NodeSnapshot>,
|
||||
store: Arc<dyn StateStore>,
|
||||
bootstrap_updates: watch::Sender<Vec<BootstrapPeer>>,
|
||||
connect_updates: mpsc::Sender<ConnectCommand>,
|
||||
trust_updates: mpsc::Sender<(PeerId, bool)>,
|
||||
}
|
||||
|
||||
impl RpcServer {
|
||||
pub(crate) async fn bind(
|
||||
address: SocketAddr,
|
||||
state: watch::Receiver<NodeSnapshot>,
|
||||
store: Arc<dyn StateStore>,
|
||||
bootstrap_updates: watch::Sender<Vec<BootstrapPeer>>,
|
||||
connect_updates: mpsc::Sender<ConnectCommand>,
|
||||
trust_updates: mpsc::Sender<(PeerId, bool)>,
|
||||
) -> Result<Self> {
|
||||
if !address.ip().is_loopback() {
|
||||
anyhow::bail!(
|
||||
"gRPC API can only bind to a loopback address until authentication is implemented"
|
||||
);
|
||||
}
|
||||
|
||||
let listener = TcpListener::bind(address)
|
||||
.await
|
||||
.with_context(|| format!("failed to bind local gRPC API at {address}"))?;
|
||||
let address = listener
|
||||
.local_addr()
|
||||
.context("failed to inspect local gRPC API address")?;
|
||||
Ok(Self {
|
||||
address,
|
||||
incoming: TcpListenerStream::new(listener),
|
||||
state,
|
||||
store,
|
||||
bootstrap_updates,
|
||||
connect_updates,
|
||||
trust_updates,
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn local_addr(&self) -> SocketAddr {
|
||||
self.address
|
||||
}
|
||||
|
||||
pub(crate) async fn serve(self, shutdown: oneshot::Receiver<()>) -> Result<()> {
|
||||
let service = NodeServiceImpl {
|
||||
state: self.state,
|
||||
store: self.store,
|
||||
bootstrap_updates: self.bootstrap_updates,
|
||||
connect_updates: self.connect_updates,
|
||||
trust_updates: self.trust_updates,
|
||||
};
|
||||
info!(address = %self.address, "local gRPC API listening");
|
||||
Server::builder()
|
||||
.add_service(proto::node_service_server::NodeServiceServer::new(service))
|
||||
.serve_with_incoming_shutdown(self.incoming, async {
|
||||
let _ = shutdown.await;
|
||||
})
|
||||
.await
|
||||
.with_context(|| format!("gRPC API server failed at {}", self.address))
|
||||
}
|
||||
}
|
||||
|
||||
struct NodeServiceImpl {
|
||||
state: watch::Receiver<NodeSnapshot>,
|
||||
store: Arc<dyn StateStore>,
|
||||
bootstrap_updates: watch::Sender<Vec<BootstrapPeer>>,
|
||||
connect_updates: mpsc::Sender<ConnectCommand>,
|
||||
trust_updates: mpsc::Sender<(PeerId, bool)>,
|
||||
}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl proto::node_service_server::NodeService for NodeServiceImpl {
|
||||
type WatchPeersStream =
|
||||
Pin<Box<dyn Stream<Item = Result<proto::ListPeersResponse, Status>> + Send + 'static>>;
|
||||
|
||||
async fn get_node_info(
|
||||
&self,
|
||||
_request: Request<proto::GetNodeInfoRequest>,
|
||||
) -> Result<Response<proto::GetNodeInfoResponse>, Status> {
|
||||
let state = self.state.borrow().clone();
|
||||
Ok(Response::new(proto::GetNodeInfoResponse {
|
||||
peer_id: state.peer_id,
|
||||
daemon_version: state.daemon_version,
|
||||
listen_addresses: state.listen_addresses,
|
||||
country_code: state.country_code.unwrap_or_default(),
|
||||
}))
|
||||
}
|
||||
|
||||
async fn list_peers(
|
||||
&self,
|
||||
_request: Request<proto::ListPeersRequest>,
|
||||
) -> Result<Response<proto::ListPeersResponse>, Status> {
|
||||
let state = self.state.borrow().clone();
|
||||
let peers = state.peers;
|
||||
Ok(Response::new(proto::ListPeersResponse {
|
||||
peers: into_proto_peers(peers),
|
||||
revision: state.peer_revision,
|
||||
}))
|
||||
}
|
||||
|
||||
async fn watch_peers(
|
||||
&self,
|
||||
_request: Request<proto::WatchPeersRequest>,
|
||||
) -> Result<Response<Self::WatchPeersStream>, Status> {
|
||||
let stream = stream::unfold(
|
||||
(self.state.clone(), None),
|
||||
|(mut state, mut last_revision): (watch::Receiver<NodeSnapshot>, Option<u64>)| async move {
|
||||
loop {
|
||||
if last_revision.is_some() && state.changed().await.is_err() {
|
||||
return None;
|
||||
}
|
||||
|
||||
let snapshot = state.borrow_and_update().clone();
|
||||
if last_revision == Some(snapshot.peer_revision) {
|
||||
continue;
|
||||
}
|
||||
last_revision = Some(snapshot.peer_revision);
|
||||
|
||||
let response = proto::ListPeersResponse {
|
||||
peers: into_proto_peers(snapshot.peers),
|
||||
revision: snapshot.peer_revision,
|
||||
};
|
||||
return Some((Ok(response), (state, last_revision)));
|
||||
}
|
||||
},
|
||||
);
|
||||
Ok(Response::new(Box::pin(stream)))
|
||||
}
|
||||
|
||||
async fn get_bootstrap_peers(
|
||||
&self,
|
||||
_request: Request<proto::GetBootstrapPeersRequest>,
|
||||
) -> Result<Response<proto::GetBootstrapPeersResponse>, Status> {
|
||||
let peers = self.store.load_bootstrap_peers().await.map_err(|error| {
|
||||
Status::internal(format!("failed to load bootstrap peers: {error}"))
|
||||
})?;
|
||||
Ok(Response::new(proto::GetBootstrapPeersResponse {
|
||||
peers: into_proto_bootstrap_peers(peers),
|
||||
}))
|
||||
}
|
||||
|
||||
async fn set_bootstrap_peers(
|
||||
&self,
|
||||
request: Request<proto::SetBootstrapPeersRequest>,
|
||||
) -> Result<Response<proto::SetBootstrapPeersResponse>, Status> {
|
||||
let peers = parse_bootstrap_peers(request.into_inner().peers)?;
|
||||
self.store
|
||||
.replace_bootstrap_peers(&peers)
|
||||
.await
|
||||
.map_err(|error| {
|
||||
Status::internal(format!("failed to save bootstrap peers: {error}"))
|
||||
})?;
|
||||
self.bootstrap_updates.send_replace(peers.clone());
|
||||
Ok(Response::new(proto::SetBootstrapPeersResponse {
|
||||
peers: into_proto_bootstrap_peers(peers),
|
||||
}))
|
||||
}
|
||||
|
||||
async fn connect_peer(
|
||||
&self,
|
||||
request: Request<proto::ConnectPeerRequest>,
|
||||
) -> Result<Response<proto::ConnectPeerResponse>, Status> {
|
||||
let address: Multiaddr = request.into_inner().address.parse().map_err(|error| {
|
||||
Status::invalid_argument(format!("invalid peer multiaddress: {error}"))
|
||||
})?;
|
||||
let peer_id = address
|
||||
.iter()
|
||||
.last()
|
||||
.and_then(|part| match part {
|
||||
Protocol::P2p(peer_id) => Some(peer_id),
|
||||
_ => None,
|
||||
})
|
||||
.ok_or_else(|| Status::invalid_argument("peer address must end with /p2p/<peer-id>"))?;
|
||||
let local_peer_id = self
|
||||
.state
|
||||
.borrow()
|
||||
.peer_id
|
||||
.parse::<PeerId>()
|
||||
.map_err(|_| Status::internal("local node has an invalid peer ID"))?;
|
||||
if peer_id == local_peer_id {
|
||||
return Err(Status::invalid_argument(
|
||||
"cannot connect to this node's own peer ID",
|
||||
));
|
||||
}
|
||||
if self
|
||||
.state
|
||||
.borrow()
|
||||
.peers
|
||||
.iter()
|
||||
.any(|peer| peer.peer_id == peer_id.to_string() && peer.connected)
|
||||
{
|
||||
return Ok(Response::new(proto::ConnectPeerResponse {
|
||||
peer_id: peer_id.to_string(),
|
||||
}));
|
||||
}
|
||||
let (reply, result) = oneshot::channel();
|
||||
self.connect_updates
|
||||
.send((peer_id, address, reply))
|
||||
.await
|
||||
.map_err(|_| {
|
||||
Status::unavailable("networking loop is not accepting peer connections")
|
||||
})?;
|
||||
result
|
||||
.await
|
||||
.map_err(|_| Status::unavailable("networking loop stopped before dialing peer"))?
|
||||
.map_err(|error| {
|
||||
Status::unavailable(format!("failed to start peer connection: {error}"))
|
||||
})?;
|
||||
Ok(Response::new(proto::ConnectPeerResponse {
|
||||
peer_id: peer_id.to_string(),
|
||||
}))
|
||||
}
|
||||
|
||||
async fn set_peer_compute_trust(
|
||||
&self,
|
||||
request: Request<proto::SetPeerComputeTrustRequest>,
|
||||
) -> Result<Response<proto::SetPeerComputeTrustResponse>, Status> {
|
||||
let request = request.into_inner();
|
||||
let peer_id: PeerId = request
|
||||
.peer_id
|
||||
.parse()
|
||||
.map_err(|error| Status::invalid_argument(format!("invalid peer ID: {error}")))?;
|
||||
self.store
|
||||
.set_peer_compute_trust(peer_id, request.trusted)
|
||||
.await
|
||||
.map_err(|error| {
|
||||
Status::internal(format!("failed to update compute trust: {error}"))
|
||||
})?;
|
||||
self.trust_updates
|
||||
.send((peer_id, request.trusted))
|
||||
.await
|
||||
.map_err(|_| Status::unavailable("networking loop is not accepting trust updates"))?;
|
||||
Ok(Response::new(proto::SetPeerComputeTrustResponse {
|
||||
trusted: request.trusted,
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_bootstrap_peers(peers: Vec<proto::BootstrapPeer>) -> Result<Vec<BootstrapPeer>, Status> {
|
||||
let mut parsed = Vec::with_capacity(peers.len());
|
||||
for peer in peers {
|
||||
let peer_id: PeerId = peer
|
||||
.peer_id
|
||||
.parse()
|
||||
.map_err(|error| Status::invalid_argument(format!("invalid peer ID: {error}")))?;
|
||||
let address: Multiaddr = peer
|
||||
.address
|
||||
.parse()
|
||||
.map_err(|error| Status::invalid_argument(format!("invalid peer address: {error}")))?;
|
||||
match address.iter().last() {
|
||||
Some(Protocol::P2p(address_peer_id)) if address_peer_id == peer_id => {}
|
||||
Some(Protocol::P2p(address_peer_id)) => {
|
||||
return Err(Status::invalid_argument(format!(
|
||||
"address peer ID {address_peer_id} does not match supplied peer ID {peer_id}"
|
||||
)));
|
||||
}
|
||||
_ => {
|
||||
return Err(Status::invalid_argument(
|
||||
"bootstrap address must end with /p2p/<peer-id>",
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
let entry = BootstrapPeer { peer_id, address };
|
||||
if !parsed.contains(&entry) {
|
||||
parsed.push(entry);
|
||||
}
|
||||
}
|
||||
Ok(parsed)
|
||||
}
|
||||
|
||||
fn into_proto_bootstrap_peers(peers: Vec<BootstrapPeer>) -> Vec<proto::BootstrapPeer> {
|
||||
peers
|
||||
.into_iter()
|
||||
.map(|peer| proto::BootstrapPeer {
|
||||
peer_id: peer.peer_id.to_string(),
|
||||
address: peer.address.to_string(),
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn into_proto_peers(peers: Vec<PeerSnapshot>) -> Vec<proto::Peer> {
|
||||
peers
|
||||
.into_iter()
|
||||
.map(|peer| proto::Peer {
|
||||
peer_id: peer.peer_id,
|
||||
connection_state: if peer.connected {
|
||||
proto::PeerConnectionState::Connected as i32
|
||||
} else {
|
||||
proto::PeerConnectionState::Disconnected as i32
|
||||
},
|
||||
known_addresses: peer.known_addresses,
|
||||
trusted_for_compute: peer.trusted_for_compute,
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{NodeServiceImpl, NodeSnapshot, PeerSnapshot, RpcServer, proto};
|
||||
use crate::storage::{BootstrapPeer, SqliteStore, StateStore};
|
||||
use libp2p::{Multiaddr, identity};
|
||||
use proto::{node_service_client::NodeServiceClient, node_service_server::NodeService};
|
||||
use std::sync::Arc;
|
||||
use tonic::Request;
|
||||
|
||||
async fn test_storage(directory: &tempfile::TempDir) -> Arc<dyn StateStore> {
|
||||
Arc::new(
|
||||
SqliteStore::open(directory.path().join("node.sqlite3"))
|
||||
.await
|
||||
.expect("open test storage"),
|
||||
)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn node_service_exposes_current_node_and_peer_snapshot() {
|
||||
let snapshot = NodeSnapshot {
|
||||
peer_id: "node-peer-id".to_owned(),
|
||||
daemon_version: "0.1.0".to_owned(),
|
||||
country_code: Some("IL".to_owned()),
|
||||
listen_addresses: vec!["/ip4/127.0.0.1/tcp/7400".to_owned()],
|
||||
peers: vec![PeerSnapshot {
|
||||
peer_id: "worker-peer-id".to_owned(),
|
||||
connected: true,
|
||||
known_addresses: vec!["/ip4/127.0.0.1/tcp/7410".to_owned()],
|
||||
trusted_for_compute: false,
|
||||
}],
|
||||
peer_revision: 0,
|
||||
};
|
||||
let (_sender, receiver) = tokio::sync::watch::channel(snapshot);
|
||||
let directory = tempfile::tempdir().expect("temporary test directory");
|
||||
let store = test_storage(&directory).await;
|
||||
let (bootstrap_updates, _bootstrap_receiver) = tokio::sync::watch::channel(Vec::new());
|
||||
let (connect_updates, _connect_receiver) = tokio::sync::mpsc::channel(4);
|
||||
let (trust_updates, _trust_receiver) = tokio::sync::mpsc::channel(4);
|
||||
let service = NodeServiceImpl {
|
||||
state: receiver,
|
||||
store,
|
||||
bootstrap_updates,
|
||||
connect_updates,
|
||||
trust_updates,
|
||||
};
|
||||
|
||||
let node = service
|
||||
.get_node_info(Request::new(proto::GetNodeInfoRequest {}))
|
||||
.await
|
||||
.expect("node info response")
|
||||
.into_inner();
|
||||
assert_eq!(node.peer_id, "node-peer-id");
|
||||
assert_eq!(node.daemon_version, "0.1.0");
|
||||
assert_eq!(node.country_code, "IL");
|
||||
|
||||
let peers = service
|
||||
.list_peers(Request::new(proto::ListPeersRequest {}))
|
||||
.await
|
||||
.expect("peer list response")
|
||||
.into_inner();
|
||||
assert_eq!(peers.peers.len(), 1);
|
||||
assert_eq!(peers.peers[0].peer_id, "worker-peer-id");
|
||||
assert_eq!(
|
||||
peers.peers[0].connection_state,
|
||||
proto::PeerConnectionState::Connected as i32
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn generated_grpc_client_can_call_node_service() {
|
||||
let local_peer_id = identity::Keypair::generate_ed25519().public().to_peer_id();
|
||||
let snapshot = NodeSnapshot {
|
||||
peer_id: local_peer_id.to_string(),
|
||||
daemon_version: "0.1.0".to_owned(),
|
||||
country_code: Some("US".to_owned()),
|
||||
listen_addresses: vec!["/ip4/127.0.0.1/tcp/7400".to_owned()],
|
||||
peers: vec![PeerSnapshot {
|
||||
peer_id: "worker-peer-id".to_owned(),
|
||||
connected: false,
|
||||
known_addresses: vec!["/ip4/127.0.0.1/tcp/7410".to_owned()],
|
||||
trusted_for_compute: false,
|
||||
}],
|
||||
peer_revision: 0,
|
||||
};
|
||||
let (sender, receiver) = tokio::sync::watch::channel(snapshot.clone());
|
||||
let directory = tempfile::tempdir().expect("temporary test directory");
|
||||
let store = test_storage(&directory).await;
|
||||
let (bootstrap_updates, mut bootstrap_receiver) = tokio::sync::watch::channel(Vec::new());
|
||||
let (connect_updates, mut connect_receiver) = tokio::sync::mpsc::channel(4);
|
||||
let (trust_updates, mut trust_receiver) = tokio::sync::mpsc::channel(4);
|
||||
let server = RpcServer::bind(
|
||||
"127.0.0.1:0".parse().expect("loopback socket address"),
|
||||
receiver,
|
||||
store.clone(),
|
||||
bootstrap_updates,
|
||||
connect_updates,
|
||||
trust_updates,
|
||||
)
|
||||
.await
|
||||
.expect("bind API server");
|
||||
let address = server.local_addr();
|
||||
let (shutdown_sender, shutdown_receiver) = tokio::sync::oneshot::channel();
|
||||
let server_task = tokio::spawn(server.serve(shutdown_receiver));
|
||||
|
||||
let mut client = NodeServiceClient::connect(format!("http://{address}"))
|
||||
.await
|
||||
.expect("connect generated gRPC client");
|
||||
let node = client
|
||||
.get_node_info(Request::new(proto::GetNodeInfoRequest {}))
|
||||
.await
|
||||
.expect("call GetNodeInfo")
|
||||
.into_inner();
|
||||
assert_eq!(node.peer_id, local_peer_id.to_string());
|
||||
assert_eq!(node.country_code, "US");
|
||||
|
||||
let peers = client
|
||||
.list_peers(Request::new(proto::ListPeersRequest {}))
|
||||
.await
|
||||
.expect("call ListPeers")
|
||||
.into_inner();
|
||||
assert_eq!(peers.peers[0].peer_id, "worker-peer-id");
|
||||
assert_eq!(
|
||||
peers.peers[0].connection_state,
|
||||
proto::PeerConnectionState::Disconnected as i32
|
||||
);
|
||||
|
||||
let mut updates = client
|
||||
.watch_peers(Request::new(proto::WatchPeersRequest {}))
|
||||
.await
|
||||
.expect("start peer watch")
|
||||
.into_inner();
|
||||
let initial = updates
|
||||
.message()
|
||||
.await
|
||||
.expect("read initial peer snapshot")
|
||||
.expect("initial peer snapshot exists");
|
||||
assert_eq!(initial.revision, 0);
|
||||
assert_eq!(
|
||||
initial.peers[0].connection_state,
|
||||
proto::PeerConnectionState::Disconnected as i32
|
||||
);
|
||||
|
||||
let mut connected_snapshot = snapshot;
|
||||
connected_snapshot.peers[0].connected = true;
|
||||
connected_snapshot.peer_revision = 1;
|
||||
sender.send_replace(connected_snapshot);
|
||||
let update = tokio::time::timeout(std::time::Duration::from_secs(1), updates.message())
|
||||
.await
|
||||
.expect("peer update arrives promptly")
|
||||
.expect("read peer update")
|
||||
.expect("peer update exists");
|
||||
assert_eq!(update.revision, 1);
|
||||
assert_eq!(
|
||||
update.peers[0].connection_state,
|
||||
proto::PeerConnectionState::Connected as i32
|
||||
);
|
||||
|
||||
let initial_bootstrap_peers = client
|
||||
.get_bootstrap_peers(Request::new(proto::GetBootstrapPeersRequest {}))
|
||||
.await
|
||||
.expect("call GetBootstrapPeers")
|
||||
.into_inner();
|
||||
assert!(initial_bootstrap_peers.peers.is_empty());
|
||||
|
||||
let direct_peer_id = identity::Keypair::generate_ed25519().public().to_peer_id();
|
||||
let direct_address = format!("/ip4/127.0.0.1/tcp/7411/p2p/{direct_peer_id}");
|
||||
let mut direct_client = client.clone();
|
||||
let connect_task = tokio::spawn(async move {
|
||||
direct_client
|
||||
.connect_peer(Request::new(proto::ConnectPeerRequest {
|
||||
address: direct_address,
|
||||
}))
|
||||
.await
|
||||
});
|
||||
let (requested_peer, requested_address, reply) = connect_receiver
|
||||
.recv()
|
||||
.await
|
||||
.expect("network receives direct connection request");
|
||||
assert_eq!(requested_peer, direct_peer_id);
|
||||
assert!(
|
||||
requested_address
|
||||
.to_string()
|
||||
.ends_with(&direct_peer_id.to_string())
|
||||
);
|
||||
reply.send(Ok(())).expect("return dial result");
|
||||
assert_eq!(
|
||||
connect_task.await.unwrap().unwrap().into_inner().peer_id,
|
||||
direct_peer_id.to_string()
|
||||
);
|
||||
|
||||
client
|
||||
.set_peer_compute_trust(Request::new(proto::SetPeerComputeTrustRequest {
|
||||
peer_id: direct_peer_id.to_string(),
|
||||
trusted: true,
|
||||
}))
|
||||
.await
|
||||
.expect("set explicit compute trust");
|
||||
assert_eq!(
|
||||
trust_receiver.recv().await,
|
||||
Some((direct_peer_id, true)),
|
||||
"network receives explicit trust update"
|
||||
);
|
||||
assert!(
|
||||
store
|
||||
.load_compute_trusted_peers()
|
||||
.await
|
||||
.unwrap()
|
||||
.contains(&direct_peer_id)
|
||||
);
|
||||
|
||||
let keypair = identity::Keypair::generate_ed25519();
|
||||
let peer_id = keypair.public().to_peer_id();
|
||||
let multiaddr: Multiaddr = format!("/ip4/127.0.0.1/tcp/7410/p2p/{peer_id}")
|
||||
.parse()
|
||||
.expect("valid bootstrap multiaddress");
|
||||
let setting = proto::BootstrapPeer {
|
||||
peer_id: peer_id.to_string(),
|
||||
address: multiaddr.to_string(),
|
||||
};
|
||||
let saved = client
|
||||
.set_bootstrap_peers(Request::new(proto::SetBootstrapPeersRequest {
|
||||
peers: vec![setting.clone()],
|
||||
}))
|
||||
.await
|
||||
.expect("call SetBootstrapPeers")
|
||||
.into_inner();
|
||||
assert_eq!(saved.peers, vec![setting.clone()]);
|
||||
bootstrap_receiver
|
||||
.changed()
|
||||
.await
|
||||
.expect("network receives bootstrap update");
|
||||
assert_eq!(
|
||||
*bootstrap_receiver.borrow_and_update(),
|
||||
vec![BootstrapPeer {
|
||||
peer_id,
|
||||
address: multiaddr.clone(),
|
||||
}]
|
||||
);
|
||||
|
||||
let loaded = client
|
||||
.get_bootstrap_peers(Request::new(proto::GetBootstrapPeersRequest {}))
|
||||
.await
|
||||
.expect("call GetBootstrapPeers")
|
||||
.into_inner();
|
||||
assert_eq!(loaded.peers, vec![setting]);
|
||||
|
||||
let mismatched_peer_id = identity::Keypair::generate_ed25519().public().to_peer_id();
|
||||
let invalid = client
|
||||
.set_bootstrap_peers(Request::new(proto::SetBootstrapPeersRequest {
|
||||
peers: vec![proto::BootstrapPeer {
|
||||
peer_id: mismatched_peer_id.to_string(),
|
||||
address: multiaddr.to_string(),
|
||||
}],
|
||||
}))
|
||||
.await
|
||||
.expect_err("reject peer ID/address mismatch");
|
||||
assert_eq!(invalid.code(), tonic::Code::InvalidArgument);
|
||||
|
||||
shutdown_sender
|
||||
.send(())
|
||||
.expect("signal gRPC server shutdown");
|
||||
drop(sender);
|
||||
tokio::time::timeout(std::time::Duration::from_secs(2), server_task)
|
||||
.await
|
||||
.expect("gRPC server shuts down promptly")
|
||||
.expect("gRPC server task joins")
|
||||
.expect("gRPC server shuts down cleanly");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn api_server_only_binds_loopback_and_reports_bind_conflicts() {
|
||||
let directory = tempfile::tempdir().expect("temporary test directory");
|
||||
let store = test_storage(&directory).await;
|
||||
let (_state_sender, state_receiver) = tokio::sync::watch::channel(NodeSnapshot::default());
|
||||
let (bootstrap_sender, _bootstrap_receiver) = tokio::sync::watch::channel(Vec::new());
|
||||
let (connect_sender, _connect_receiver) = tokio::sync::mpsc::channel(4);
|
||||
let (trust_sender, _trust_receiver) = tokio::sync::mpsc::channel(4);
|
||||
|
||||
let non_loopback = RpcServer::bind(
|
||||
"0.0.0.0:50051".parse().expect("socket address"),
|
||||
state_receiver.clone(),
|
||||
store.clone(),
|
||||
bootstrap_sender.clone(),
|
||||
connect_sender.clone(),
|
||||
trust_sender.clone(),
|
||||
)
|
||||
.await
|
||||
.err()
|
||||
.expect("non-loopback bind must be rejected");
|
||||
assert!(non_loopback.to_string().contains("loopback"));
|
||||
|
||||
let occupied_listener = tokio::net::TcpListener::bind("127.0.0.1:0")
|
||||
.await
|
||||
.expect("bind occupied socket");
|
||||
let occupied_address = occupied_listener.local_addr().expect("occupied address");
|
||||
let bind_error = RpcServer::bind(
|
||||
occupied_address,
|
||||
state_receiver,
|
||||
store,
|
||||
bootstrap_sender,
|
||||
connect_sender,
|
||||
trust_sender,
|
||||
)
|
||||
.await
|
||||
.err()
|
||||
.expect("conflicting bind must return an error");
|
||||
assert!(
|
||||
bind_error
|
||||
.to_string()
|
||||
.contains("failed to bind local gRPC API")
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
mod sqlite;
|
||||
|
||||
use std::{path::PathBuf, sync::Arc};
|
||||
|
||||
use anyhow::Result;
|
||||
use async_trait::async_trait;
|
||||
use libp2p::{Multiaddr, PeerId, identity};
|
||||
|
||||
use crate::platform;
|
||||
|
||||
pub(crate) use sqlite::SqliteStore;
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub(crate) struct BootstrapPeer {
|
||||
pub(crate) peer_id: PeerId,
|
||||
pub(crate) address: Multiaddr,
|
||||
}
|
||||
|
||||
/// Durable, node-local state. Network peers never share this store.
|
||||
#[async_trait]
|
||||
pub(crate) trait StateStore: Send + Sync {
|
||||
async fn load_or_create_identity(&self) -> Result<identity::Keypair>;
|
||||
async fn clear_bootstrap_peers(&self) -> Result<()>;
|
||||
async fn save_bootstrap_peer(&self, peer: &BootstrapPeer) -> Result<()>;
|
||||
async fn replace_bootstrap_peers(&self, peers: &[BootstrapPeer]) -> Result<()>;
|
||||
async fn load_bootstrap_peers(&self) -> Result<Vec<BootstrapPeer>>;
|
||||
async fn set_peer_compute_trust(&self, peer_id: PeerId, trusted: bool) -> Result<()>;
|
||||
async fn load_compute_trusted_peers(&self) -> Result<Vec<PeerId>>;
|
||||
}
|
||||
|
||||
pub(crate) async fn open(data_dir: Option<PathBuf>) -> Result<Arc<dyn StateStore>> {
|
||||
let database_path = platform::database_path(data_dir)?;
|
||||
Ok(Arc::new(SqliteStore::open(database_path).await?))
|
||||
}
|
||||
@@ -0,0 +1,400 @@
|
||||
use std::{
|
||||
fs::{self, OpenOptions},
|
||||
path::{Path, PathBuf},
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use anyhow::{Context, Result, bail};
|
||||
use async_trait::async_trait;
|
||||
use libp2p::identity;
|
||||
use sqlx::{
|
||||
Row,
|
||||
sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous},
|
||||
};
|
||||
use tracing::info;
|
||||
|
||||
use super::{BootstrapPeer, StateStore};
|
||||
|
||||
static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");
|
||||
|
||||
pub(crate) struct SqliteStore {
|
||||
pool: sqlx::SqlitePool,
|
||||
}
|
||||
|
||||
impl SqliteStore {
|
||||
pub(crate) async fn open(database_path: PathBuf) -> Result<Self> {
|
||||
prepare_database_path(&database_path)?;
|
||||
|
||||
let options = SqliteConnectOptions::new()
|
||||
.filename(&database_path)
|
||||
.create_if_missing(true)
|
||||
.foreign_keys(true)
|
||||
.journal_mode(SqliteJournalMode::Wal)
|
||||
.synchronous(SqliteSynchronous::Normal)
|
||||
.busy_timeout(Duration::from_secs(5));
|
||||
let pool = SqlitePoolOptions::new()
|
||||
.max_connections(4)
|
||||
.connect_with(options)
|
||||
.await
|
||||
.with_context(|| {
|
||||
format!(
|
||||
"failed to open node database at {}",
|
||||
database_path.display()
|
||||
)
|
||||
})?;
|
||||
|
||||
MIGRATOR.run(&pool).await.with_context(|| {
|
||||
format!(
|
||||
"failed to migrate node database at {}",
|
||||
database_path.display()
|
||||
)
|
||||
})?;
|
||||
|
||||
info!(database_path = %database_path.display(), "opened node state database");
|
||||
Ok(Self { pool })
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl StateStore for SqliteStore {
|
||||
async fn load_or_create_identity(&self) -> Result<identity::Keypair> {
|
||||
if let Some(row) = sqlx::query(
|
||||
"SELECT encoding_version, private_key FROM node_identity WHERE singleton_id = 1",
|
||||
)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?
|
||||
{
|
||||
return decode_identity(row.get::<i64, _>(0), row.get::<Vec<u8>, _>(1));
|
||||
}
|
||||
|
||||
let candidate = identity::Keypair::generate_ed25519();
|
||||
let encoded = candidate.to_protobuf_encoding()?;
|
||||
sqlx::query(
|
||||
"INSERT OR IGNORE INTO node_identity (singleton_id, encoding_version, private_key) VALUES (1, 1, ?)",
|
||||
)
|
||||
.bind(encoded)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.context("failed to persist node identity")?;
|
||||
|
||||
let row = sqlx::query(
|
||||
"SELECT encoding_version, private_key FROM node_identity WHERE singleton_id = 1",
|
||||
)
|
||||
.fetch_one(&self.pool)
|
||||
.await
|
||||
.context("node identity was not present after initialization")?;
|
||||
|
||||
decode_identity(row.get::<i64, _>(0), row.get::<Vec<u8>, _>(1))
|
||||
}
|
||||
|
||||
async fn save_bootstrap_peer(&self, peer: &BootstrapPeer) -> Result<()> {
|
||||
sqlx::query("INSERT OR IGNORE INTO bootstrap_peers (peer_id, address) VALUES (?, ?)")
|
||||
.bind(peer.peer_id.to_string())
|
||||
.bind(peer.address.to_string())
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.context("failed to persist bootstrap peer")?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn replace_bootstrap_peers(&self, peers: &[BootstrapPeer]) -> Result<()> {
|
||||
let mut transaction = self
|
||||
.pool
|
||||
.begin()
|
||||
.await
|
||||
.context("failed to start bootstrap peer update")?;
|
||||
sqlx::query("DELETE FROM bootstrap_peers")
|
||||
.execute(&mut *transaction)
|
||||
.await
|
||||
.context("failed to clear existing bootstrap peers")?;
|
||||
|
||||
for peer in peers {
|
||||
sqlx::query("INSERT OR IGNORE INTO bootstrap_peers (peer_id, address) VALUES (?, ?)")
|
||||
.bind(peer.peer_id.to_string())
|
||||
.bind(peer.address.to_string())
|
||||
.execute(&mut *transaction)
|
||||
.await
|
||||
.context("failed to save bootstrap peer")?;
|
||||
}
|
||||
|
||||
transaction
|
||||
.commit()
|
||||
.await
|
||||
.context("failed to commit bootstrap peer update")
|
||||
}
|
||||
|
||||
async fn clear_bootstrap_peers(&self) -> Result<()> {
|
||||
sqlx::query("DELETE FROM bootstrap_peers")
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.context("failed to clear bootstrap peers")?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn load_bootstrap_peers(&self) -> Result<Vec<BootstrapPeer>> {
|
||||
let rows =
|
||||
sqlx::query("SELECT peer_id, address FROM bootstrap_peers ORDER BY peer_id, address")
|
||||
.fetch_all(&self.pool)
|
||||
.await
|
||||
.context("failed to load bootstrap peers")?;
|
||||
|
||||
rows.into_iter()
|
||||
.map(|row| {
|
||||
let peer_id_text: String = row.get(0);
|
||||
let address_text: String = row.get(1);
|
||||
let peer_id = peer_id_text.parse().with_context(|| {
|
||||
format!("invalid peer ID in bootstrap peer store: {peer_id_text}")
|
||||
})?;
|
||||
let address = address_text.parse().with_context(|| {
|
||||
format!("invalid multiaddress in bootstrap peer store: {address_text}")
|
||||
})?;
|
||||
Ok(BootstrapPeer { peer_id, address })
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
async fn set_peer_compute_trust(&self, peer_id: libp2p::PeerId, trusted: bool) -> Result<()> {
|
||||
if trusted {
|
||||
sqlx::query("INSERT OR IGNORE INTO trusted_compute_peers (peer_id) VALUES (?)")
|
||||
.bind(peer_id.to_string())
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.context("failed to trust peer for compute")?;
|
||||
} else {
|
||||
sqlx::query("DELETE FROM trusted_compute_peers WHERE peer_id = ?")
|
||||
.bind(peer_id.to_string())
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.context("failed to revoke compute trust for peer")?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn load_compute_trusted_peers(&self) -> Result<Vec<libp2p::PeerId>> {
|
||||
let peer_ids: Vec<String> =
|
||||
sqlx::query_scalar("SELECT peer_id FROM trusted_compute_peers ORDER BY peer_id")
|
||||
.fetch_all(&self.pool)
|
||||
.await
|
||||
.context("failed to load compute trusted peers")?;
|
||||
peer_ids
|
||||
.into_iter()
|
||||
.map(|peer_id| {
|
||||
peer_id
|
||||
.parse()
|
||||
.with_context(|| format!("invalid peer ID in compute trust store: {peer_id}"))
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
fn decode_identity(encoding_version: i64, encoded: Vec<u8>) -> Result<identity::Keypair> {
|
||||
if encoding_version != 1 {
|
||||
bail!("unsupported node identity encoding version: {encoding_version}");
|
||||
}
|
||||
|
||||
identity::Keypair::from_protobuf_encoding(&encoded)
|
||||
.context("stored node identity is not a valid libp2p key")
|
||||
}
|
||||
|
||||
fn prepare_database_path(database_path: &Path) -> Result<()> {
|
||||
let data_dir = database_path
|
||||
.parent()
|
||||
.context("node database path must have a parent directory")?;
|
||||
fs::create_dir_all(data_dir).with_context(|| {
|
||||
format!(
|
||||
"failed to create node data directory {}",
|
||||
data_dir.display()
|
||||
)
|
||||
})?;
|
||||
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
|
||||
|
||||
fs::set_permissions(data_dir, fs::Permissions::from_mode(0o700))
|
||||
.with_context(|| format!("failed to restrict permissions on {}", data_dir.display()))?;
|
||||
|
||||
OpenOptions::new()
|
||||
.write(true)
|
||||
.create(true)
|
||||
.truncate(false)
|
||||
.mode(0o600)
|
||||
.open(database_path)
|
||||
.with_context(|| {
|
||||
format!("failed to create node database {}", database_path.display())
|
||||
})?;
|
||||
fs::set_permissions(database_path, fs::Permissions::from_mode(0o600)).with_context(
|
||||
|| {
|
||||
format!(
|
||||
"failed to restrict permissions on {}",
|
||||
database_path.display()
|
||||
)
|
||||
},
|
||||
)?;
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
{
|
||||
OpenOptions::new()
|
||||
.write(true)
|
||||
.create(true)
|
||||
.truncate(false)
|
||||
.open(database_path)
|
||||
.with_context(|| {
|
||||
format!("failed to create node database {}", database_path.display())
|
||||
})?;
|
||||
// On Windows, the database inherits the current user's directory ACL.
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::storage::BootstrapPeer;
|
||||
|
||||
use super::{SqliteStore, StateStore};
|
||||
use libp2p::{Multiaddr, identity};
|
||||
|
||||
#[tokio::test]
|
||||
async fn identity_survives_store_reopen() {
|
||||
let directory = tempfile::tempdir().expect("temporary data directory");
|
||||
let database_path = directory.path().join("node.sqlite3");
|
||||
|
||||
let first_peer_id = {
|
||||
let store = SqliteStore::open(database_path.clone())
|
||||
.await
|
||||
.expect("open initial store");
|
||||
store
|
||||
.load_or_create_identity()
|
||||
.await
|
||||
.expect("create identity")
|
||||
.public()
|
||||
.to_peer_id()
|
||||
};
|
||||
|
||||
let store = SqliteStore::open(database_path)
|
||||
.await
|
||||
.expect("reopen store");
|
||||
let second_peer_id = store
|
||||
.load_or_create_identity()
|
||||
.await
|
||||
.expect("load persisted identity")
|
||||
.public()
|
||||
.to_peer_id();
|
||||
|
||||
assert_eq!(first_peer_id, second_peer_id);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn concurrent_initialization_keeps_one_identity() {
|
||||
let directory = tempfile::tempdir().expect("temporary data directory");
|
||||
let database_path = directory.path().join("node.sqlite3");
|
||||
let store_a = SqliteStore::open(database_path.clone())
|
||||
.await
|
||||
.expect("open first store");
|
||||
let store_b = SqliteStore::open(database_path)
|
||||
.await
|
||||
.expect("open second store");
|
||||
|
||||
let (identity_a, identity_b) = tokio::join!(
|
||||
store_a.load_or_create_identity(),
|
||||
store_b.load_or_create_identity()
|
||||
);
|
||||
let peer_a = identity_a.expect("first identity").public().to_peer_id();
|
||||
let peer_b = identity_b.expect("second identity").public().to_peer_id();
|
||||
|
||||
assert_eq!(peer_a, peer_b);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn bootstrap_peers_survive_store_reopen_and_are_deduplicated() {
|
||||
let directory = tempfile::tempdir().expect("temporary data directory");
|
||||
let database_path = directory.path().join("node.sqlite3");
|
||||
let keypair = identity::Keypair::generate_ed25519();
|
||||
let peer_id = keypair.public().to_peer_id();
|
||||
let address: Multiaddr = "/ip4/127.0.0.1/tcp/7400"
|
||||
.parse()
|
||||
.expect("valid test address");
|
||||
|
||||
let store = SqliteStore::open(database_path.clone())
|
||||
.await
|
||||
.expect("open store");
|
||||
store
|
||||
.save_bootstrap_peer(&BootstrapPeer {
|
||||
peer_id,
|
||||
address: address.clone(),
|
||||
})
|
||||
.await
|
||||
.expect("save bootstrap peer");
|
||||
store
|
||||
.save_bootstrap_peer(&BootstrapPeer {
|
||||
peer_id,
|
||||
address: address.clone(),
|
||||
})
|
||||
.await
|
||||
.expect("save duplicate bootstrap peer");
|
||||
drop(store);
|
||||
|
||||
let reopened = SqliteStore::open(database_path)
|
||||
.await
|
||||
.expect("reopen store");
|
||||
let peers = reopened
|
||||
.load_bootstrap_peers()
|
||||
.await
|
||||
.expect("load bootstrap peers");
|
||||
|
||||
assert_eq!(peers, vec![BootstrapPeer { peer_id, address }]);
|
||||
|
||||
reopened
|
||||
.clear_bootstrap_peers()
|
||||
.await
|
||||
.expect("clear bootstrap peers");
|
||||
assert!(reopened.load_bootstrap_peers().await.unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn encoded_identity_is_a_private_keypair() {
|
||||
let keypair = identity::Keypair::generate_ed25519();
|
||||
let encoded = keypair.to_protobuf_encoding().expect("encode identity");
|
||||
let decoded = identity::Keypair::from_protobuf_encoding(&encoded).expect("decode identity");
|
||||
|
||||
assert_eq!(keypair.public().to_peer_id(), decoded.public().to_peer_id());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn compute_trust_survives_reopen_and_can_be_revoked() {
|
||||
let directory = tempfile::tempdir().expect("temporary data directory");
|
||||
let database_path = directory.path().join("node.sqlite3");
|
||||
let peer_id = identity::Keypair::generate_ed25519().public().to_peer_id();
|
||||
let store = SqliteStore::open(database_path.clone())
|
||||
.await
|
||||
.expect("open store");
|
||||
|
||||
store
|
||||
.set_peer_compute_trust(peer_id, true)
|
||||
.await
|
||||
.expect("trust peer");
|
||||
let reopened = SqliteStore::open(database_path)
|
||||
.await
|
||||
.expect("reopen store");
|
||||
assert_eq!(
|
||||
reopened.load_compute_trusted_peers().await.unwrap(),
|
||||
vec![peer_id]
|
||||
);
|
||||
|
||||
reopened
|
||||
.set_peer_compute_trust(peer_id, false)
|
||||
.await
|
||||
.expect("revoke trust");
|
||||
assert!(
|
||||
reopened
|
||||
.load_compute_trusted_peers()
|
||||
.await
|
||||
.unwrap()
|
||||
.is_empty()
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -1 +0,0 @@
|
||||
|
||||
Generated
+27
@@ -0,0 +1,27 @@
|
||||
{
|
||||
"nodes": {
|
||||
"nixpkgs": {
|
||||
"locked": {
|
||||
"lastModified": 1791048980,
|
||||
"narHash": "sha256-KgItSKML8Xvte0B7/uGnBDsYzSnnKHcOaiUbgWBXLXw=",
|
||||
"owner": "NixOS",
|
||||
"repo": "nixpkgs",
|
||||
"rev": "a7868a727837f3c09cee2ce0ca671c76b1589fed",
|
||||
"type": "github"
|
||||
},
|
||||
"original": {
|
||||
"owner": "NixOS",
|
||||
"ref": "nixos-unstable",
|
||||
"repo": "nixpkgs",
|
||||
"type": "github"
|
||||
}
|
||||
},
|
||||
"root": {
|
||||
"inputs": {
|
||||
"nixpkgs": "nixpkgs"
|
||||
}
|
||||
}
|
||||
},
|
||||
"root": "root",
|
||||
"version": 7
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
{
|
||||
description = "Sisyphus Rust P2P development environment";
|
||||
|
||||
inputs.nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
|
||||
|
||||
outputs = { nixpkgs, ... }:
|
||||
let
|
||||
systems = [ "x86_64-linux" "aarch64-linux" "x86_64-darwin" "aarch64-darwin" ];
|
||||
forAllSystems = function:
|
||||
builtins.listToAttrs (map (system: {
|
||||
name = system;
|
||||
value = function system;
|
||||
}) systems);
|
||||
in
|
||||
{
|
||||
devShells = forAllSystems (system:
|
||||
let pkgs = import nixpkgs { inherit system; };
|
||||
in {
|
||||
default = pkgs.mkShell {
|
||||
packages = with pkgs; [
|
||||
rustc
|
||||
cargo
|
||||
rustfmt
|
||||
clippy
|
||||
rust-analyzer
|
||||
protobuf
|
||||
pkg-config
|
||||
clang
|
||||
cmake
|
||||
cargo-watch
|
||||
cargo-deny
|
||||
cargo-audit
|
||||
nodejs_22
|
||||
gtk3
|
||||
gsettings-desktop-schemas
|
||||
];
|
||||
|
||||
shellHook = ''
|
||||
export XDG_DATA_DIRS="${pkgs.gtk3}/share/gsettings-schemas/${pkgs.gtk3.name}:${pkgs.gsettings-desktop-schemas}/share/gsettings-schemas/${pkgs.gsettings-desktop-schemas.name}:$XDG_DATA_DIRS"
|
||||
echo "Sisyphus development shell (Rust + Electron/React)"
|
||||
echo "Rust: $(rustc --version)"
|
||||
echo "Protobuf: $(protoc --version)"
|
||||
echo "Node: $(node --version)"
|
||||
'';
|
||||
};
|
||||
});
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
syntax = "proto3";
|
||||
|
||||
package sisyphus.node.v1;
|
||||
|
||||
service NodeService {
|
||||
rpc GetNodeInfo(GetNodeInfoRequest) returns (GetNodeInfoResponse);
|
||||
rpc ListPeers(ListPeersRequest) returns (ListPeersResponse);
|
||||
rpc WatchPeers(WatchPeersRequest) returns (stream ListPeersResponse);
|
||||
rpc GetBootstrapPeers(GetBootstrapPeersRequest) returns (GetBootstrapPeersResponse);
|
||||
rpc SetBootstrapPeers(SetBootstrapPeersRequest) returns (SetBootstrapPeersResponse);
|
||||
rpc ConnectPeer(ConnectPeerRequest) returns (ConnectPeerResponse);
|
||||
rpc SetPeerComputeTrust(SetPeerComputeTrustRequest) returns (SetPeerComputeTrustResponse);
|
||||
}
|
||||
|
||||
message GetNodeInfoRequest {}
|
||||
|
||||
message GetNodeInfoResponse {
|
||||
string peer_id = 1;
|
||||
string daemon_version = 2;
|
||||
repeated string listen_addresses = 3;
|
||||
string country_code = 4;
|
||||
}
|
||||
|
||||
message ListPeersRequest {}
|
||||
|
||||
message WatchPeersRequest {}
|
||||
|
||||
message GetBootstrapPeersRequest {}
|
||||
|
||||
message GetBootstrapPeersResponse {
|
||||
repeated BootstrapPeer peers = 1;
|
||||
}
|
||||
|
||||
message SetBootstrapPeersRequest {
|
||||
repeated BootstrapPeer peers = 1;
|
||||
}
|
||||
|
||||
message SetBootstrapPeersResponse {
|
||||
repeated BootstrapPeer peers = 1;
|
||||
}
|
||||
|
||||
message ConnectPeerRequest { string address = 1; }
|
||||
message ConnectPeerResponse { string peer_id = 1; }
|
||||
message SetPeerComputeTrustRequest {
|
||||
string peer_id = 1;
|
||||
bool trusted = 2;
|
||||
}
|
||||
message SetPeerComputeTrustResponse { bool trusted = 1; }
|
||||
|
||||
message BootstrapPeer {
|
||||
string peer_id = 1;
|
||||
string address = 2;
|
||||
}
|
||||
|
||||
message ListPeersResponse {
|
||||
repeated Peer peers = 1;
|
||||
uint64 revision = 2;
|
||||
}
|
||||
|
||||
message Peer {
|
||||
string peer_id = 1;
|
||||
PeerConnectionState connection_state = 2;
|
||||
repeated string known_addresses = 3;
|
||||
bool trusted_for_compute = 4;
|
||||
}
|
||||
|
||||
enum PeerConnectionState {
|
||||
PEER_CONNECTION_STATE_UNSPECIFIED = 0;
|
||||
PEER_CONNECTION_STATE_DISCONNECTED = 1;
|
||||
PEER_CONNECTION_STATE_CONNECTED = 2;
|
||||
}
|
||||
Reference in New Issue
Block a user