plana 0.2.1

PLANA protocol foundation: the canonical JSON-RPC 2.0 wire layer (unix-socket / HTTP / WS bindings, typed method routing) merged with the generic protocol core (handshake, identity, RBAC, regions, health descriptors) — the base every plana-rpc-* and domain-profile crate builds on.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

use anyhow::{Context, Result};
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
use std::path::Path;

use interprocess::local_socket::{
    prelude::*,
    {GenericFilePath, ListenerOptions, Stream as LocalSocketStream},
};
pub use interprocess::local_socket::{
    traits::Listener as InterprocessAccept, Listener as InterprocessListener,
    ListenerNonblockingMode, Stream as InterprocessStream,
};
use tracing::{info, warn};

const SOCKET_PERMS: u32 = 0o777;

pub fn ensure_socket_dir(socket_path: &Path) -> Result<()> {
    if let Some(parent) = socket_path.parent() {
        std::fs::create_dir_all(parent)
            .with_context(|| format!("failed to create socket dir {}", parent.display()))?;

        chmod_recursive(parent, SOCKET_PERMS);
    }
    Ok(())
}

pub fn remove_stale_socket(socket_path: &Path) {
    if socket_path.exists() {
        if let Err(e) = std::fs::remove_file(socket_path) {
            warn!(
                path = %socket_path.display(),
                error = %e,
                "failed to remove stale socket file"
            );
        }
    }
}

pub fn chmod_socket(socket_path: &Path) {
    chmod_recursive(socket_path, SOCKET_PERMS);
}

#[cfg(unix)]
pub async fn bind_tokio(socket_path: &Path) -> Result<tokio::net::UnixListener> {
    ensure_socket_dir(socket_path)?;
    remove_stale_socket(socket_path);

    let listener = tokio::net::UnixListener::bind(socket_path)
        .with_context(|| format!("failed to bind Unix socket {}", socket_path.display()))?;

    chmod_socket(socket_path);

    info!(path = %socket_path.display(), "Unix socket bound (tokio)");
    Ok(listener)
}

#[cfg(unix)]
pub async fn connect_tokio(socket_path: &Path) -> Result<tokio::net::UnixStream> {
    let stream = tokio::net::UnixStream::connect(socket_path)
        .await
        .with_context(|| format!("failed to connect to Unix socket {}", socket_path.display()))?;
    info!(path = %socket_path.display(), "Connected to Unix socket (tokio)");
    Ok(stream)
}

pub fn bind_interprocess(socket_path: &Path) -> Result<interprocess::local_socket::Listener> {
    ensure_socket_dir(socket_path)?;
    remove_stale_socket(socket_path);

    let name = socket_path
        .to_fs_name::<GenericFilePath>()
        .with_context(|| {
            format!(
                "failed to convert path to socket name: {}",
                socket_path.display()
            )
        })?;

    let listener = ListenerOptions::new()
        .name(name)
        .create_sync()
        .with_context(|| {
            format!(
                "failed to create LocalSocket listener at {}",
                socket_path.display()
            )
        })?;

    chmod_socket(socket_path);

    info!(path = %socket_path.display(), "Unix socket bound (interprocess)");
    Ok(listener)
}

pub fn connect_interprocess(socket_path: &Path) -> Result<interprocess::local_socket::Stream> {
    let name = socket_path
        .to_fs_name::<GenericFilePath>()
        .with_context(|| {
            format!(
                "failed to convert path to socket name: {}",
                socket_path.display()
            )
        })?;

    let stream = LocalSocketStream::connect(name)
        .with_context(|| format!("failed to connect to {}", socket_path.display()))?;

    info!(path = %socket_path.display(), "Connected to Unix socket (interprocess)");
    Ok(stream)
}

#[cfg(unix)]
fn chmod_recursive(path: &Path, mode: u32) {
    if let Ok(metadata) = std::fs::metadata(path) {
        let mut perms = metadata.permissions();
        perms.set_mode(mode);
        if let Err(e) = std::fs::set_permissions(path, perms) {
            warn!(
                path = %path.display(),
                error = %e,
                mode = format!("{:#o}", mode),
                "failed to set permissions"
            );
        }
    }
}

#[cfg(not(unix))]
fn chmod_recursive(_path: &Path, _mode: u32) {}