KofTwentyTwo/nuncio
2
0 Fork 0
Code Issues 73 Pull Requests 12 Actions Packages Projects Releases Wiki Activity

refactor: remove the legacy JSON-RPC IPC; daemon serves gRPC only

Browse Source 
Delete nuncio-core::ipc (IpcDaemonServer, IpcClient, framing, protocol -- the
demux-race transport), the daemon's :9422 boot + JSON-RPC handler, the stubbed
'nuncio-cli daemon' subcommand, and the IPC-transport test (gRPC coverage
already exists). The daemon now awaits nunciod::grpc::serve as its foreground
task; DB/keyring/sync/outbox/update remain background tasks.

Refs #173
This commit is contained in:
James Maes committed 2026-07-27 07:45:06 -05:00
1 parent 32a7f7c788
commit 6ead680c12
13 files changed
+36 -1069
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/nuncio-cli/src/args.rs
-6
View File
@@ -132,12 +132,6 @@ pub enum Commands {
#[command(subcommand)]
action: ContactSubcommand,
},
/// Launch centralized local background server daemon (`nuncio daemon`).
Daemon {
/// TCP port to bind IPC daemon server to (default: 9422).
#[arg(long, default_value = "9422")]
port: u16,
},
}
/// Contact subcommands (`nuncio contact <verb>`).
crates/nuncio-cli/src/runner.rs
-8
View File
@@ -677,14 +677,6 @@ impl HeadlessRunner {
}
},
},
Commands::Daemon { port } => {
let addr = format!("127.0.0.1:{}", port);
if json_mode {
format_json(&json!({ "status": "daemon_running", "bind_addr": addr }))
} else {
format!("Nuncio IPC Daemon listening on {}", addr)
}
}
}
}
crates/nuncio-core/src/ipc/client.rs
-225
View File
@@ -1,225 +0,0 @@
//! IPC Client used by presentation shells to connect to running `nunciod` daemon.
use crate::ipc::framing::{read_frame, write_frame};
use crate::ipc::protocol::{JsonRpcNotification, JsonRpcRequest, JsonRpcResponse};
use crate::CoreCommand;
use serde_json::{json, Value};
use std::process::{Command, Stdio};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use thiserror::Error;
use tokio::net::TcpStream;
use tokio::sync::mpsc;
/// Errors emitted by `IpcClient`.
#[derive(Error, Debug)]
pub enum IpcClientError {
/// Transport I/O failure.
#[error("I/O transport error: {0}")]
Io(#[from] std::io::Error),
/// JSON serialization error.
#[error("JSON serialization error: {0}")]
Json(#[from] serde_json::Error),
/// Failed to spawn or connect to background `nunciod` daemon after max retries.
#[error("Failed to connect or spawn daemon after max retries")]
SpawnFailed,
/// RPC execution error returned by server.
#[error("RPC error ({code}): {message}")]
Rpc {
/// RPC error code.
code: i32,
/// RPC error message.
message: String,
},
}
/// Client endpoint managing communication with `IpcDaemonServer`.
#[derive(Clone)]
pub struct IpcClient {
server_addr: String,
request_id_counter: std::sync::Arc<AtomicU64>,
}
impl IpcClient {
/// Create a new `IpcClient` pointing to target server address.
pub fn new(server_addr: impl Into<String>) -> Self {
Self {
server_addr: server_addr.into(),
request_id_counter: std::sync::Arc::new(AtomicU64::new(1)),
}
}
/// Connect to target socket or transparently spawn background `nunciod` process.
pub async fn connect_or_spawn(&self) -> Result<TcpStream, IpcClientError> {
if let Ok(stream) = TcpStream::connect(&self.server_addr).await {
return Ok(stream);
}
let _ = self.spawn_daemon();
let mut backoff = Duration::from_millis(50);
for _ in 0..5 {
tokio::time::sleep(backoff).await;
if let Ok(stream) = TcpStream::connect(&self.server_addr).await {
return Ok(stream);
}
backoff *= 2;
}
Err(IpcClientError::SpawnFailed)
}
/// Spawn `nunciod` daemon in background.
fn spawn_daemon(&self) -> Result<(), IpcClientError> {
let exe_path = std::env::current_exe()?
.parent()
.map(|p| p.join("nuncio-cli"))
.unwrap_or_else(|| std::path::PathBuf::from("nuncio-cli"));
let mut cmd = Command::new(exe_path);
cmd.arg("daemon");
cmd.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null());
#[cfg(windows)]
{
use std::os::windows::process::CommandExt;
const CREATE_NO_WINDOW: u32 = 0x08000000;
cmd.creation_flags(CREATE_NO_WINDOW);
}
let _ = cmd.spawn();
Ok(())
}
/// Execute a JSON-RPC request against running daemon and return result payload.
pub async fn call_rpc(&self, method: &str, params: Value) -> Result<Value, IpcClientError> {
let mut stream = self.connect_or_spawn().await?;
let req_id = self.request_id_counter.fetch_add(1, Ordering::SeqCst);
let req = JsonRpcRequest::new(req_id, method, params);
let payload = serde_json::to_vec(&req)?;
write_frame(&mut stream, &payload).await?;
let resp_bytes = read_frame(&mut stream).await?;
let resp: JsonRpcResponse = serde_json::from_slice(&resp_bytes)?;
if let Some(err) = resp.error {
return Err(IpcClientError::Rpc {
code: err.code,
message: err.message,
});
}
Ok(resp.result.unwrap_or(Value::Null))
}
/// Execute a `ping` health check against running daemon.
pub async fn ping(&self) -> Result<bool, IpcClientError> {
let result = self.call_rpc("system.ping", json!({})).await?;
Ok(result.as_str() == Some("pong"))
}
/// Dispatch a core domain command to running daemon.
pub async fn send_command(&self, cmd: CoreCommand) -> Result<Value, IpcClientError> {
match cmd {
CoreCommand::SyncAll => self.call_rpc("mail.sync_all", json!({})).await,
CoreCommand::MarkRead { message_id, read } => {
self.call_rpc(
"mail.mark_read",
json!({ "message_id": message_id, "read": read }),
)
.await
}
_ => self.call_rpc("mail.sync_all", json!({})).await,
}
}
/// Fetch application state snapshot from daemon.
pub async fn get_state(&self) -> Result<Value, IpcClientError> {
self.call_rpc("system.state", json!({})).await
}
/// List filter rules from running daemon.
pub async fn filter_list(&self) -> Result<Value, IpcClientError> {
self.call_rpc("filter.list", json!({})).await
}
/// Create a new filter rule in running daemon.
pub async fn filter_create(
&self,
name: &str,
nsql: &str,
priority: i32,
) -> Result<Value, IpcClientError> {
self.call_rpc(
"filter.create",
json!({ "name": name, "nsql": nsql, "priority": priority }),
)
.await
}
/// Edit an existing filter rule in running daemon.
pub async fn filter_edit(
&self,
id: &str,
name: &str,
nsql: &str,
priority: i32,
) -> Result<Value, IpcClientError> {
self.call_rpc(
"filter.edit",
json!({ "id": id, "name": name, "nsql": nsql, "priority": priority }),
)
.await
}
/// Delete a filter rule in running daemon.
pub async fn filter_delete(&self, id: &str) -> Result<Value, IpcClientError> {
self.call_rpc("filter.delete", json!({ "id": id })).await
}
/// Dry-run preview evaluation of an email against filter rules.
pub async fn filter_preview(
&self,
email: &crate::model::Email,
) -> Result<Value, IpcClientError> {
let email_json = serde_json::to_value(email)?;
self.call_rpc("filter.preview", json!({ "email": email_json }))
.await
}
/// Query filter execution logs from running daemon.
pub async fn filter_logs(&self, limit: usize) -> Result<Value, IpcClientError> {
self.call_rpc("filter.logs", json!({ "limit": limit }))
.await
}
/// Subscribe to real-time `events.notify` server push stream.
pub async fn subscribe_events(
&self,
) -> Result<mpsc::Receiver<JsonRpcNotification>, IpcClientError> {
let stream = self.connect_or_spawn().await?;
let (tx, rx) = mpsc::channel(100);
tokio::spawn(async move {
let (mut reader, _writer) = tokio::io::split(stream);
loop {
let frame_bytes = match read_frame(&mut reader).await {
Ok(bytes) => bytes,
Err(_) => break,
};
if let Ok(notification) =
serde_json::from_slice::<JsonRpcNotification>(&frame_bytes)
{
if tx.send(notification).await.is_err() {
break;
}
}
}
});
Ok(rx)
}
}
crates/nuncio-core/src/ipc/framing.rs
-81
View File
@@ -1,81 +0,0 @@
//! Length-prefixed binary frame codec for Nuncio IPC protocol.
use std::io::Error as IoError;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
/// Maximum allowed payload size per frame (16 MB).
pub const MAX_FRAME_SIZE: usize = 16 * 1024 * 1024;
/// Read a 4-byte big-endian length-prefixed payload frame from an async reader.
pub async fn read_frame<R>(reader: &mut R) -> Result<Vec<u8>, IoError>
where
R: AsyncRead + Unpin,
{
let mut len_buf = [0u8; 4];
reader.read_exact(&mut len_buf).await?;
let length = u32::from_be_bytes(len_buf) as usize;
if length > MAX_FRAME_SIZE {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("Frame payload size {length} exceeds maximum limit of {MAX_FRAME_SIZE} bytes"),
));
}
let mut payload = vec![0u8; length];
reader.read_exact(&mut payload).await?;
Ok(payload)
}
/// Write a 4-byte big-endian length-prefixed payload frame to an async writer.
pub async fn write_frame<W>(writer: &mut W, payload: &[u8]) -> Result<(), IoError>
where
W: AsyncWrite + Unpin,
{
if payload.len() > MAX_FRAME_SIZE {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
format!(
"Payload size {} exceeds maximum limit of {} bytes",
payload.len(),
MAX_FRAME_SIZE
),
));
}
let len_bytes = (payload.len() as u32).to_be_bytes();
writer.write_all(&len_bytes).await?;
writer.write_all(payload).await?;
writer.flush().await?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use tokio::io::duplex;
#[tokio::test]
async fn frame_codec_roundtrip_success() {
let (mut client, mut server) = duplex(1024);
let test_payload = b"{\"jsonrpc\":\"2.0\",\"method\":\"ping\"}";
let write_task = tokio::spawn(async move {
write_frame(&mut client, test_payload).await.unwrap();
});
let read_payload = read_frame(&mut server).await.unwrap();
write_task.await.unwrap();
assert_eq!(&read_payload[..], test_payload);
}
#[tokio::test]
async fn frame_codec_oversized_payload_fails() {
let (mut client, _server) = duplex(1024);
let oversized = vec![0u8; MAX_FRAME_SIZE + 1];
let result = write_frame(&mut client, &oversized).await;
assert!(result.is_err());
}
}
crates/nuncio-core/src/ipc/mod.rs
-13
View File
@@ -1,13 +0,0 @@
//! Local IPC Client-Server Daemon Engine for Nuncio.
//! Provides centralized JSON-RPC 2.0 length-prefixed communication over local sockets so that
//! CLI, TUI, GUI, and MCP shells all talk to the SAME running `nunciod` background server.
pub mod client;
pub mod framing;
pub mod protocol;
pub mod server;
pub use client::{IpcClient, IpcClientError};
pub use framing::{read_frame, write_frame, MAX_FRAME_SIZE};
pub use protocol::{JsonRpcError, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse};
pub use server::IpcDaemonServer;
crates/nuncio-core/src/ipc/protocol.rs
-126
View File
@@ -1,126 +0,0 @@
//! JSON-RPC 2.0 data models and wire contracts for Nuncio IPC.
use serde::{Deserialize, Serialize};
use serde_json::Value;
/// JSON-RPC 2.0 Request Frame.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct JsonRpcRequest {
/// Protocol version string (must be `"2.0"`).
pub jsonrpc: String,
/// Numeric request identifier.
pub id: u64,
/// RPC method name (e.g. `system.ping`, `mail.sync_all`).
pub method: String,
/// Optional parameter payload.
#[serde(default)]
pub params: Value,
}
impl JsonRpcRequest {
/// Construct a new JSON-RPC 2.0 request frame.
pub fn new(id: u64, method: impl Into<String>, params: Value) -> Self {
Self {
jsonrpc: "2.0".to_string(),
id,
method: method.into(),
params,
}
}
}
/// JSON-RPC 2.0 Response Frame.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct JsonRpcResponse {
/// Protocol version string (`"2.0"`).
pub jsonrpc: String,
/// Numeric request identifier matching corresponding request.
pub id: u64,
/// Result payload if successful.
#[serde(skip_serializing_if = "Option::is_none")]
pub result: Option<Value>,
/// Error payload if execution failed.
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<JsonRpcError>,
}
impl JsonRpcResponse {
/// Construct a successful response frame.
pub fn success(id: u64, result: Value) -> Self {
Self {
jsonrpc: "2.0".to_string(),
id,
result: Some(result),
error: None,
}
}
/// Construct an error response frame.
pub fn error(id: u64, code: i32, message: impl Into<String>) -> Self {
Self {
jsonrpc: "2.0".to_string(),
id,
result: None,
error: Some(JsonRpcError {
code,
message: message.into(),
data: None,
}),
}
}
}
/// JSON-RPC 2.0 Error Detail.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct JsonRpcError {
/// Numeric error code (-32600 to -32603, or domain-specific integer).
pub code: i32,
/// Descriptive error message string.
pub message: String,
/// Optional additional error data context.
#[serde(skip_serializing_if = "Option::is_none")]
pub data: Option<Value>,
}
/// JSON-RPC 2.0 Unidirectional Notification Frame (Server Push).
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct JsonRpcNotification {
/// Protocol version string (`"2.0"`).
pub jsonrpc: String,
/// Notification event topic (e.g. `events.notify`).
pub method: String,
/// Event payload parameters.
pub params: Value,
}
impl JsonRpcNotification {
/// Construct a new JSON-RPC 2.0 notification frame.
pub fn new(method: impl Into<String>, params: Value) -> Self {
Self {
jsonrpc: "2.0".to_string(),
method: method.into(),
params,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn json_rpc_request_response_serde_roundtrip() {
let req = JsonRpcRequest::new(42, "system.ping", json!({ "client": "tui" }));
let serialized = serde_json::to_string(&req).expect("serialize request");
let deserialized: JsonRpcRequest =
serde_json::from_str(&serialized).expect("deserialize request");
assert_eq!(req, deserialized);
let resp = JsonRpcResponse::success(42, json!("pong"));
let resp_serialized = serde_json::to_string(&resp).expect("serialize response");
let resp_deserialized: JsonRpcResponse =
serde_json::from_str(&resp_serialized).expect("deserialize response");
assert_eq!(resp, resp_deserialized);
}
}
crates/nuncio-core/src/ipc/server.rs
-220
View File
@@ -1,220 +0,0 @@
//! Local IPC Daemon Server handling multi-client socket communication over `EventBus`.
use crate::ipc::framing::{read_frame, write_frame};
use crate::ipc::protocol::{JsonRpcNotification, JsonRpcRequest, JsonRpcResponse};
use crate::{CoreCommand, EventBus};
use futures::future::BoxFuture;
use serde_json::json;
use std::sync::Arc;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::net::TcpListener;
/// Async handler callback signature for custom RPC extensions (e.g. `filter.*` handlers).
pub type CustomRpcHandler = Arc<
dyn Fn(&str, serde_json::Value) -> BoxFuture<'static, Option<Result<serde_json::Value, String>>>
+ Send
+ Sync,
>;
/// Centralized IPC Daemon Server managing presentation shell clients over `EventBus`.
pub struct IpcDaemonServer {
event_bus: Arc<EventBus>,
bind_addr: String,
custom_handler: Option<CustomRpcHandler>,
}
impl IpcDaemonServer {
/// Default TCP loopback bind address (`127.0.0.1:9422`).
pub const DEFAULT_ADDR: &'static str = "127.0.0.1:9422";
/// Create a new `IpcDaemonServer` wrapping shared `EventBus`.
pub fn new(event_bus: Arc<EventBus>, bind_addr: impl Into<String>) -> Self {
Self {
event_bus,
bind_addr: bind_addr.into(),
custom_handler: None,
}
}
/// Create a new `IpcDaemonServer` with custom RPC handler extension.
pub fn with_handler(
event_bus: Arc<EventBus>,
bind_addr: impl Into<String>,
handler: CustomRpcHandler,
) -> Self {
Self {
event_bus,
bind_addr: bind_addr.into(),
custom_handler: Some(handler),
}
}
/// Attach a custom RPC handler callback.
pub fn set_handler(&mut self, handler: CustomRpcHandler) {
self.custom_handler = Some(handler);
}
/// Return configured bind address.
pub fn bind_addr(&self) -> &str {
&self.bind_addr
}
/// Start listening for client socket connections and processing JSON-RPC requests.
pub async fn run_server(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let listener = TcpListener::bind(&self.bind_addr).await?;
tracing::info!("Nuncio IPC Daemon listening on {}", self.bind_addr);
loop {
let (stream, _addr) = listener.accept().await?;
let event_bus = self.event_bus.clone();
let handler = self.custom_handler.clone();
tokio::spawn(async move {
let _ = Self::handle_stream_with_handler(stream, event_bus, handler).await;
});
}
}
/// Handle stream without custom handler.
pub async fn handle_stream<S>(
stream: S,
event_bus: Arc<EventBus>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>>
where
S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
Self::handle_stream_with_handler(stream, event_bus, None).await
}
/// Handle bi-directional request processing and event streaming over an async byte stream.
pub async fn handle_stream_with_handler<S>(
stream: S,
event_bus: Arc<EventBus>,
custom_handler: Option<CustomRpcHandler>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>>
where
S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
let (mut reader, mut writer) = tokio::io::split(stream);
let (push_tx, mut push_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(100);
// Task to write push notifications and RPC responses to stream
let writer_task = tokio::spawn(async move {
while let Some(bytes) = push_rx.recv().await {
if write_frame(&mut writer, &bytes).await.is_err() {
break;
}
}
});
// Event listener broadcasting CoreEvents over push channel
let mut core_event_rx = event_bus.subscribe_events();
let push_tx_events = push_tx.clone();
let event_listener_task = tokio::spawn(async move {
while let Ok(evt) = core_event_rx.recv().await {
let notification = JsonRpcNotification::new(
"events.notify",
json!({ "event": format!("{:?}", evt) }),
);
if let Ok(bytes) = serde_json::to_vec(&notification) {
if push_tx_events.send(bytes).await.is_err() {
break;
}
}
}
});
// Read loop processing RPC requests from client
loop {
let frame_bytes = match read_frame(&mut reader).await {
Ok(bytes) => bytes,
Err(_) => break,
};
let req: JsonRpcRequest = match serde_json::from_slice(&frame_bytes) {
Ok(r) => r,
Err(_) => continue,
};
let response = match req.method.as_str() {
"system.ping" => JsonRpcResponse::success(req.id, json!("pong")),
"system.state" => {
let state = event_bus.current_state();
JsonRpcResponse::success(
req.id,
json!({
"status": format!("{:?}", state.status),
"accounts_loaded": state.accounts_loaded,
"unread_count": state.unread_count,
"last_error": state.last_error,
}),
)
}
"mail.sync_all" => {
// Flip status to `Syncing` and publish `SyncStarted`
// synchronously so a caller polling `system.state` right
// after this RPC returns observes it deterministically.
event_bus.process_command(CoreCommand::SyncAll);
// ALSO enqueue the command onto the async command
// channel so whichever real `CoreCommand` consumer owns
// `take_command_receiver()` (e.g. `nunciod`'s real
// inbound-sync command loop) performs the actual
// fetch/persist work and later flips status back to
// `Idle` + publishes
// `SyncCompleted`. Best-effort: if no consumer is
// currently running (some test harnesses only exercise
// this transport in isolation), the command is simply
// queued and never drained, which is harmless -- the
// synchronous status flip above already satisfies this
// RPC's own contract.
let _ = event_bus.send_command(CoreCommand::SyncAll).await;
JsonRpcResponse::success(req.id, json!({ "status": "dispatched" }))
}
"mail.mark_read" => {
if let Some(msg_id) = req.params.get("message_id").and_then(|v| v.as_str()) {
let read = req
.params
.get("read")
.and_then(|v| v.as_bool())
.unwrap_or(true);
self_mark_read(&event_bus, msg_id, read);
JsonRpcResponse::success(req.id, json!({ "status": "marked" }))
} else {
JsonRpcResponse::error(req.id, -32602, "missing message_id")
}
}
other_method => {
if let Some(handler) = &custom_handler {
if let Some(res) = handler(other_method, req.params.clone()).await {
match res {
Ok(val) => JsonRpcResponse::success(req.id, val),
Err(err_msg) => JsonRpcResponse::error(req.id, -32603, err_msg),
}
} else {
JsonRpcResponse::error(req.id, -32601, "Method not found")
}
} else {
JsonRpcResponse::error(req.id, -32601, "Method not found")
}
}
};
if let Ok(resp_bytes) = serde_json::to_vec(&response) {
if push_tx.send(resp_bytes).await.is_err() {
break;
}
}
}
event_listener_task.abort();
writer_task.abort();
Ok(())
}
}
fn self_mark_read(event_bus: &EventBus, message_id: &str, read: bool) {
event_bus.process_command(CoreCommand::MarkRead {
message_id: message_id.to_string(),
read,
});
}
crates/nuncio-core/src/lib.rs
-1
View File
@@ -7,7 +7,6 @@ pub mod ai;
pub mod config;
pub mod e2ee;
pub mod export;
pub mod ipc;
pub mod mcp_policy;
pub mod model;
pub mod plugin;
crates/nuncio-core/tests/ipc_system_test.rs
-48
View File
@@ -1,48 +0,0 @@
//! System integration test suite for nuncio-core IPC client-server framing & JSON-RPC protocol.
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
use nuncio_core::ipc::{IpcClient, IpcDaemonServer};
use nuncio_core::{CoreCommand, EventBus};
use std::sync::Arc;
#[tokio::test]
async fn system_test_ipc_daemon_ping_state_and_commands() {
let event_bus = Arc::new(EventBus::new());
let addr = "127.0.0.1:19423";
let server = IpcDaemonServer::new(event_bus.clone(), addr);
tokio::spawn(async move {
let _ = server.run_server().await;
});
let client = IpcClient::new(addr);
// Give server time to bind TCP socket
let mut ping_res = false;
for _ in 0..15 {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
if let Ok(res) = client.ping().await {
ping_res = res;
if ping_res {
break;
}
}
}
assert!(ping_res);
// 2. Fetch State
let state_res = client.get_state().await.expect("get state success");
assert_eq!(state_res["status"].as_str().unwrap(), "Idle");
// 3. Dispatch CoreCommand over IPC
let sync_res = client
.send_command(CoreCommand::SyncAll)
.await
.expect("send command success");
assert_eq!(sync_res["status"], "dispatched");
assert_eq!(
event_bus.current_state().status,
nuncio_core::EngineStatus::Syncing
);
}
crates/nuncio-proto/src/addr.rs
+1 -2
View File
@@ -6,8 +6,7 @@
//! endpoint without depending on the `nunciod` binary crate.
/// Default loopback bind address for the `nuncio.v1` gRPC server, used when
/// the [`GRPC_ADDR_ENV_VAR`] environment variable is unset. Deliberately
/// distinct from the legacy JSON-RPC IPC server's default (`127.0.0.1:9422`).
/// the [`GRPC_ADDR_ENV_VAR`] environment variable is unset.
pub const DEFAULT_GRPC_ADDR: &str = "127.0.0.1:9420";
/// Environment variable overriding the gRPC bind/connect address.
crates/nunciod/src/grpc.rs
+1 -3
View File
@@ -3,9 +3,7 @@
//! token sourced from the OS keyring vault
//! (`nuncio_store::vault::GRPC_TOKEN_ACCOUNT`).
//!
//! This runs ALONGSIDE the existing hand-rolled JSON-RPC IPC server
//! (`nuncio_core::ipc::IpcDaemonServer`); migrating callers off the JSON-RPC
//! transport is out of scope here.
//! This is the daemon's sole client-facing transport.
use nuncio_core::{CoreCommand, CoreEvent, EventBus};
use nuncio_filter::{FilterEngine, NsqlParser, NsqlValidator, ValidationOptions};
crates/nunciod/src/main.rs
+25 -231
View File
@@ -1,12 +1,9 @@
//! Centralized Standalone Background Daemon Server Binary (`nunciod`).
//! Owns storage persistence, background sync loops, protocol connections,
//! filter automation engine, outbox retries, and multi-client IPC socket distribution.
//! filter automation engine, outbox retries, and the multi-client gRPC API.
use nuncio_core::ipc::server::CustomRpcHandler;
use nuncio_core::ipc::IpcDaemonServer;
use nuncio_core::{CoreCommand, CoreEvent, EventBus};
use nuncio_filter::{FilterEngine, NsqlParser, NsqlValidator, OutboxManager, ValidationOptions};
use serde_json::json;
use nuncio_filter::{FilterEngine, OutboxManager};
use std::sync::Arc;
#[tokio::main]
@@ -172,206 +169,16 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
None
};
// Custom RPC Handler for filter.* and update.* methods
let db_rpc = db.clone();
let engine_rpc = filter_engine.clone();
let event_bus_rpc = event_bus.clone();
let handler: CustomRpcHandler = Arc::new(move |method, params| {
let db = db_rpc.clone();
let engine = engine_rpc.clone();
let event_bus = event_bus_rpc.clone();
let method_str = method.to_string();
Box::pin(async move {
match method_str.as_str() {
"filter.list" => match db.list_filter_rules().await {
Ok(rules) => Some(Ok(json!(rules))),
Err(e) => Some(Err(e.to_string())),
},
"filter.create" => {
let name = params
.get("name")
.and_then(|v| v.as_str())
.unwrap_or("Untitled Rule");
let nsql = params.get("nsql").and_then(|v| v.as_str()).unwrap_or("");
let priority =
params.get("priority").and_then(|v| v.as_i64()).unwrap_or(0) as i32;
match NsqlParser::parse_rule(name, priority, nsql) {
Ok(rule) => {
let val_opts = ValidationOptions::default();
if let Err(val_err) = NsqlValidator::validate(&rule, &val_opts) {
return Some(Err(val_err.to_string()));
}
if let Err(e) = db.save_filter_rule(&rule).await {
return Some(Err(e.to_string()));
}
if let Ok(all_rules) = db.list_filter_rules().await {
let _ = engine.reload_rules(all_rules);
}
Some(Ok(json!(rule)))
}
Err(parse_err) => Some(Err(parse_err.to_string())),
}
}
"filter.edit" => {
let id = match params.get("id").and_then(|v| v.as_str()) {
Some(id) => id,
None => return Some(Err("missing rule id".to_string())),
};
let name = params
.get("name")
.and_then(|v| v.as_str())
.unwrap_or("Updated Rule");
let nsql = params.get("nsql").and_then(|v| v.as_str()).unwrap_or("");
let priority =
params.get("priority").and_then(|v| v.as_i64()).unwrap_or(0) as i32;
match NsqlParser::parse_rule(name, priority, nsql) {
Ok(mut rule) => {
rule.id = id.to_string();
let val_opts = ValidationOptions::default();
if let Err(val_err) = NsqlValidator::validate(&rule, &val_opts) {
return Some(Err(val_err.to_string()));
}
if let Err(e) = db.save_filter_rule(&rule).await {
return Some(Err(e.to_string()));
}
if let Ok(all_rules) = db.list_filter_rules().await {
let _ = engine.reload_rules(all_rules);
}
Some(Ok(json!(rule)))
}
Err(parse_err) => Some(Err(parse_err.to_string())),
}
}
"filter.delete" => {
let id = match params.get("id").and_then(|v| v.as_str()) {
Some(id) => id,
None => return Some(Err("missing rule id".to_string())),
};
if let Err(e) = db.delete_filter_rule(id).await {
return Some(Err(e.to_string()));
}
if let Ok(all_rules) = db.list_filter_rules().await {
let _ = engine.reload_rules(all_rules);
}
Some(Ok(json!({ "status": "deleted" })))
}
"filter.preview" => {
if let Some(email_val) = params.get("email") {
if let Ok(email) =
serde_json::from_value::<nuncio_core::model::Email>(email_val.clone())
{
let preview = engine.preview(&email);
return Some(Ok(json!(preview)));
}
}
Some(Err("invalid email payload for preview".to_string()))
}
"filter.logs" => {
let limit = params.get("limit").and_then(|v| v.as_u64()).unwrap_or(50) as usize;
match db.list_filter_execution_logs(limit).await {
Ok(logs) => Some(Ok(json!(logs))),
Err(e) => Some(Err(e.to_string())),
}
}
"filter.triage_keyset" => {
// Keyset Chunking Triage Engine
let batch_size = params
.get("batch_size")
.and_then(|v| v.as_u64())
.unwrap_or(1000) as usize;
let mut last_id = String::new();
let mut processed = 0;
let mut matched_count = 0;
loop {
let chunk = match db.get_message_chunk(&last_id, batch_size).await {
Ok(c) => c,
Err(e) => return Some(Err(e.to_string())),
};
let Some(last) = chunk.last() else {
break;
};
last_id = last.id.clone();
for email in &chunk {
processed += 1;
let matches = engine.evaluate(email);
for (rule, actions) in matches {
matched_count += 1;
for action in actions {
let action_str = action.to_nsql();
let _ = db
.save_filter_execution_log(&rule.id, &email.id, &action_str)
.await;
let outbox_item = OutboxManager::create_mutation(
&rule.id,
&email.id,
&action_str,
None,
);
let _ = db.save_pending_mutation(&outbox_item).await;
event_bus.publish_event(CoreEvent::FilterExecuted {
rule_id: rule.id.clone(),
message_id: email.id.clone(),
action_taken: action_str,
});
}
}
}
event_bus.publish_event(CoreEvent::BatchFilterProgress {
processed,
total: processed,
matched: matched_count,
});
}
Some(Ok(
json!({ "processed": processed, "matched": matched_count }),
))
}
"update.check" => match nuncio_core::UpdateEngine::new() {
Ok(updater) => match updater.check_for_updates().await {
Ok(res) => Some(Ok(json!(res))),
Err(e) => Some(Err(e.to_string())),
},
Err(e) => Some(Err(e.to_string())),
},
"update.apply" => {
// SECURITY: `UpdateEngine`'s checksum verification is
// currently fail-open (an update installs unverified if
// `SHA256SUMS.txt` is missing from the release). Until
// that checksum-verification gap is closed, nunciod must
// never perform an on-demand install either. Use
// `update.check` for a read-only version check, and
// install updates manually in the meantime.
Some(Err(
"auto-update is temporarily disabled: update verification is not yet \
safe; install updates manually for now"
.to_string(),
))
}
_ => None,
}
})
});
// gRPC `nuncio.v1.System` + `nuncio.v1.Accounts` server.
//
// Runs ALONGSIDE the existing JSON-RPC IPC server below; the migration
// off the hand-rolled JSON-RPC transport happens incrementally. The
// bearer token is minted (or loaded, on subsequent runs) from the real
// OS keyring vault via the shared `account_secrets` `SecretManager`
// (same instance the real-sync command loop above uses), fails closed
// if the keyring is unavailable, and is never logged. `Accounts` reuses
// this SAME `SecretManager` to write account password credentials to
// the OS keyring (never to SQLite) -- see `nunciod::grpc`'s security
// comment on why EVERY mounted service shares this one
// `BearerAuthInterceptor`.
// gRPC `nuncio.v1.System`, `nuncio.v1.Accounts`, `nuncio.v1.Mail`,
// `nuncio.v1.Filters`, `nuncio.v1.Export`, and `nuncio.v1.Audit` server.
// This is the daemon's ONLY client-facing transport: the bearer token is
// minted (or loaded, on subsequent runs) from the real OS keyring vault
// via the shared `account_secrets` `SecretManager` (same instance the
// real-sync command loop above uses), fails closed if the keyring is
// unavailable, and is never logged. `Accounts` reuses this SAME
// `SecretManager` to write account password credentials to the OS
// keyring (never to SQLite) -- see `nunciod::grpc`'s security comment on
// why EVERY mounted service shares this one `BearerAuthInterceptor`.
let grpc_secrets = account_secrets.clone();
let grpc_token_bytes = grpc_secrets
.get_or_create_key_bytes(nuncio_store::vault::GRPC_TOKEN_ACCOUNT, 32)
@@ -383,33 +190,20 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
nuncio.v1.Export, nuncio.v1.Audit) starting on {} (loopback only)",
grpc_addr
);
let grpc_event_bus = event_bus.clone();
let grpc_db = db.clone();
// The gRPC `Filters` service shares this SAME `filter_engine` instance
// with the JSON-RPC `filter.*` handler
// above, so a rule created/deleted through EITHER transport reloads the
// one live `ArcSwap`-backed rule set both transports evaluate against.
let grpc_filter_engine = filter_engine.clone();
let _grpc_task = tokio::spawn(async move {
if let Err(e) = nunciod::grpc::serve(
&grpc_addr,
grpc_event_bus,
grpc_db,
grpc_filter_engine,
grpc_secrets,
grpc_token,
)
.await
{
tracing::error!("nunciod gRPC server failed: {}", e);
}
});
let addr = std::env::var("NUNCIO_IPC_ADDR").unwrap_or_else(|_| "127.0.0.1:9422".to_string());
let server = IpcDaemonServer::with_handler(event_bus.clone(), &addr, handler);
tracing::info!("nunciod listening on {}", addr);
server.run_server().await?;
// The gRPC server is the long-running foreground task that keeps this
// process alive; every other subsystem above is a background worker
// spawned on top of it.
nunciod::grpc::serve(
&grpc_addr,
event_bus,
db,
filter_engine,
grpc_secrets,
grpc_token,
)
.await
.map_err(|e| format!("nunciod gRPC server failed: {e}"))?;
Ok(())
}
crates/nunciod/tests/daemon_e2e_test.rs
+9 -105
View File
@@ -1,18 +1,16 @@
//! End-to-End multi-shell daemon integration test suite.
//! End-to-End daemon integration test suite.
//!
//! Exercises the `nunciod` daemon end-to-end over the real IPC socket using
//! `nuncio-core`'s `IpcClient`/`IpcDaemonServer`, and independently exercises
//! the in-workspace `nuncio-cli` presentation shell (`HeadlessRunner`) to
//! confirm a real reference client shell continues to function alongside
//! the daemon. This suite intentionally has no dependency on the archived
//! presentation shells that were moved to `_reference/` (nuncio-tui,
//! nuncio-gui, nuncio-mcp) as part of shrinking the workspace to the
//! engine + daemon + reference CLI.
//! Boots a real `nunciod` gRPC server and drives it via the in-workspace
//! `nuncio-cli` presentation shell (`HeadlessRunner`) to confirm a genuine
//! reference client round-trips full commands against a live daemon. This
//! suite intentionally has no dependency on the archived presentation
//! shells that were moved to `_reference/` (nuncio-tui, nuncio-gui,
//! nuncio-mcp) as part of shrinking the workspace to the engine + daemon +
//! reference CLI.
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
use nuncio_cli::{AccountSubcommand, Commands, HeadlessRunner, SystemSubcommand};
use nuncio_core::ipc::{IpcClient, IpcDaemonServer};
use nuncio_cli::{Commands, HeadlessRunner, SystemSubcommand};
use nuncio_core::{CoreCommand, EventBus};
use nuncio_filter::FilterEngine;
use nuncio_store::db::DatabaseEngine;
@@ -20,100 +18,6 @@ use nuncio_store::vault::{SecretManager, GRPC_TOKEN_ACCOUNT};
use std::sync::Arc;
use tokio::net::TcpListener;
/// Boots a real `nunciod` `IpcDaemonServer` and drives it concurrently from
/// three independent `IpcClient` socket connections (standing in for any
/// number of simultaneous thin presentation shells talking JSON-RPC over
/// TCP), then separately drives the real `nuncio-cli` `HeadlessRunner`
/// reference shell to confirm it still executes full noun+verb commands
/// correctly. This preserves the daemon-boot + real-client-round-trip
/// coverage the previous multi-shell test provided, without requiring the
/// archived `nuncio-mcp` crate.
#[tokio::test]
async fn e2e_multi_shell_daemon_concurrency_test() {
let event_bus = Arc::new(EventBus::new());
let addr = "127.0.0.1:19424";
let server = IpcDaemonServer::new(event_bus.clone(), addr);
tokio::spawn(async move {
let _ = server.run_server().await;
});
// Simulate 3 concurrent thin-shell connections to the nunciod daemon
// over the real IPC socket.
let client_a = IpcClient::new(addr);
let client_b = IpcClient::new(addr);
let client_c = IpcClient::new(addr);
// Give server time to bind TCP socket.
let mut ping_res = false;
for _ in 0..15 {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
if let Ok(res) = client_a.ping().await {
ping_res = res;
if ping_res {
break;
}
}
}
assert!(ping_res);
assert!(client_b.ping().await.expect("shell b ping"));
assert!(client_c.ping().await.expect("shell c ping"));
// 2. Shell B issues a sync command over its own concurrent connection.
let sync_res = client_b
.send_command(CoreCommand::SyncAll)
.await
.expect("shell b sync command");
assert_eq!(sync_res["status"], "dispatched");
// 3. Shell A fetches state and verifies status is Syncing.
let state_res = client_a.get_state().await.expect("shell a get state");
assert_eq!(state_res["status"].as_str().unwrap(), "Syncing");
// 4. Shell C marks a message read over its own concurrent connection.
let mark_res = client_c
.send_command(CoreCommand::MarkRead {
message_id: "msg-1".to_string(),
read: true,
})
.await
.expect("shell c mark read");
assert_eq!(mark_res["status"], "marked");
// 5. Independently, drive the real in-workspace `nuncio-cli` reference
// shell (its own ephemeral local engine, per its documented headless
// design) to confirm a genuine presentation-shell client still
// round-trips full commands correctly end-to-end. `Account::Show`
// reads this runner's own ephemeral local `db` (out of scope for any
// gRPC-backed command), so it is safe to exercise without a live
// `nunciod` gRPC daemon. `mail sync`, `mail list/read/mark/search/send`,
// and `account add/list` are now real gRPC clients of the daemon and
// are exercised against a real `nunciod` gRPC server by `nuncio-cli`'s
// own test suite and by `spine_e2e_test.rs`'s full offline spine E2E,
// rather than here.
let cli_runner = HeadlessRunner::ephemeral()
.await
.expect("cli headless runner initializes");
let cli_show = cli_runner
.execute_command(
&Commands::Account {
action: AccountSubcommand::Show {
id: "missing".to_string(),
},
},
false,
)
.await;
assert!(cli_show.contains("Account 'missing' not found"));
// `system status` no longer reflects this `HeadlessRunner`'s own local
// ephemeral state: it is a real gRPC client of the `nunciod` daemon.
// That round trip is exercised in full, against a real `nunciod` gRPC
// server, by `cli_system_status_round_trips_over_grpc_to_live_daemon`
// below.
}
/// Reference-client proof that boots a real `nunciod` `nuncio.v1.System`
/// gRPC server (via
/// [`nunciod::grpc::serve_on_listener`]) on an ephemeral loopback port,