Sockets: running actors get TCP, UDP, and DNS through wasi:sockets; candidates get no network
This commit is contained in:
+4
-3
@@ -6,7 +6,8 @@ when this file disagrees with them, they win.
|
|||||||
## The shape
|
## The shape
|
||||||
|
|
||||||
- **Actors are WASI commands, not components with interfaces.** stdin,
|
- **Actors are WASI commands, not components with interfaces.** stdin,
|
||||||
stdout, files, and HTTP are enough, and every language can do them.
|
stdout, files, HTTP, and sockets are enough, and every language can do
|
||||||
|
them.
|
||||||
- **Messages are JSON lines through a broker, fire and forget.** Like
|
- **Messages are JSON lines through a broker, fire and forget.** Like
|
||||||
Erlang: no acknowledgments, no request/response. A reply is a message with
|
Erlang: no acknowledgments, no request/response. A reply is a message with
|
||||||
`reply_to`. Sausage is not a reliable message bus.
|
`reply_to`. Sausage is not a reliable message bus.
|
||||||
@@ -27,7 +28,7 @@ actor, or a change that the agent makes to itself.
|
|||||||
tests before an install, and the rollback of a version that crash-loops.
|
tests before an install, and the rollback of a version that crash-loops.
|
||||||
Without that rollback, a broken self-upgrade would end the agent.
|
Without that rollback, a broken self-upgrade would end the agent.
|
||||||
- **No permissions, no limits, no hiding.** Any actor can message, upgrade,
|
- **No permissions, no limits, no hiding.** Any actor can message, upgrade,
|
||||||
or stop any actor, read every folder, and use HTTP. People who want rules
|
or stop any actor, read every folder, and use the network. People who want rules
|
||||||
can build them.
|
can build them.
|
||||||
- **No re-delivery and no state snapshots.** A crash loses the message, and
|
- **No re-delivery and no state snapshots.** A crash loses the message, and
|
||||||
all versions of an actor share `data/` and `public/`. A rollback restores
|
all versions of an actor share `data/` and `public/`. A rollback restores
|
||||||
@@ -45,7 +46,7 @@ actor, or a change that the agent makes to itself.
|
|||||||
then. `versions/` keeps every version that was installed, so a rollback
|
then. `versions/` keeps every version that was installed, so a rollback
|
||||||
can always go back. A file that fails stays where it was put.
|
can always go back. A file that fails stays where it was put.
|
||||||
- **Tests are JSON transcripts** next to the source. They run on a copy of
|
- **Tests are JSON transcripts** next to the source. They run on a copy of
|
||||||
the actor's state, without HTTP, so a test cannot send real requests.
|
the actor's state, without network, so a test cannot send real requests.
|
||||||
- **`broker` and `cargo` never answer a reply,** so no two actors answer
|
- **`broker` and `cargo` never answer a reply,** so no two actors answer
|
||||||
each other forever. The agent starts a turn only for text that is not a
|
each other forever. The agent starts a turn only for text that is not a
|
||||||
reply, for the same reason.
|
reply, for the same reason.
|
||||||
|
|||||||
+4
-2
@@ -16,7 +16,9 @@ The instance gets:
|
|||||||
- **stderr:** free text, appended to `/actor/<id>/stderr.log`
|
- **stderr:** free text, appended to `/actor/<id>/stderr.log`
|
||||||
- **environment:** `SAUSAGE_ACTOR` (its ID), `SAUSAGE_VERSION` (its version),
|
- **environment:** `SAUSAGE_ACTOR` (its ID), `SAUSAGE_VERSION` (its version),
|
||||||
and `SAUSAGE_CANDIDATE=1` during a test run
|
and `SAUSAGE_CANDIDATE=1` during a test run
|
||||||
- **outgoing HTTP and HTTPS** through `wasi:http` (no raw sockets)
|
- **the network:** outgoing HTTP and HTTPS through `wasi:http` (the host does
|
||||||
|
TLS), and TCP, UDP, and DNS through `wasi:sockets`, including listening on
|
||||||
|
a port of the host
|
||||||
- **the filesystem** described below
|
- **the filesystem** described below
|
||||||
|
|
||||||
## Messages
|
## Messages
|
||||||
@@ -105,7 +107,7 @@ with the reason. All versions share the actor's `data/` and `public/`: a
|
|||||||
new version must read the data of the old one.
|
new version must read the data of the old one.
|
||||||
|
|
||||||
**Tests.** The broker runs the candidate on a copy of the actor's `data/`
|
**Tests.** The broker runs the candidate on a copy of the actor's `data/`
|
||||||
and `public/`, without HTTP. The tests are JSON files in
|
and `public/`, without network. The tests are JSON files in
|
||||||
`<source>/tests/*.json`, one scenario per file:
|
`<source>/tests/*.json`, one scenario per file:
|
||||||
|
|
||||||
```json
|
```json
|
||||||
|
|||||||
+2
-2
@@ -176,7 +176,7 @@ impl Broker {
|
|||||||
|
|
||||||
/// What an instance of `id` gets (see PROTOCOL.md). A candidate runs in
|
/// What an instance of `id` gets (see PROTOCOL.md). A candidate runs in
|
||||||
/// `scratch`, which holds a copy of the actor's `data/` and `public/`,
|
/// `scratch`, which holds a copy of the actor's `data/` and `public/`,
|
||||||
/// and has no HTTP.
|
/// and has no network.
|
||||||
pub fn instance_config(&self, id: &str, version: u64, scratch: Option<&Path>) -> InstanceConfig {
|
pub fn instance_config(&self, id: &str, version: u64, scratch: Option<&Path>) -> InstanceConfig {
|
||||||
let own = scratch.map_or_else(|| self.world.actor_dir(id), Path::to_path_buf);
|
let own = scratch.map_or_else(|| self.world.actor_dir(id), Path::to_path_buf);
|
||||||
let mut env = vec![("SAUSAGE_ACTOR".into(), id.to_string()), ("SAUSAGE_VERSION".into(), version.to_string())];
|
let mut env = vec![("SAUSAGE_ACTOR".into(), id.to_string()), ("SAUSAGE_VERSION".into(), version.to_string())];
|
||||||
@@ -189,7 +189,7 @@ impl Broker {
|
|||||||
Mount { guest: "/public".into(), host: own.join("public"), writable: true },
|
Mount { guest: "/public".into(), host: own.join("public"), writable: true },
|
||||||
Mount { guest: "/data".into(), host: own.join("data"), writable: true },
|
Mount { guest: "/data".into(), host: own.join("data"), writable: true },
|
||||||
],
|
],
|
||||||
allow_http: scratch.is_none(),
|
network: scratch.is_none(),
|
||||||
stderr_log: own.join("stderr.log"),
|
stderr_log: own.join("stderr.log"),
|
||||||
env,
|
env,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -18,7 +18,7 @@
|
|||||||
//! also go to that recipient.
|
//! also go to that recipient.
|
||||||
//!
|
//!
|
||||||
//! Each file runs in a fresh instance on a copy of the actor's `/data` and
|
//! Each file runs in a fresh instance on a copy of the actor's `/data` and
|
||||||
//! `/public` (empty folders for a new actor), without HTTP. A version
|
//! `/public` (empty folders for a new actor), without network. A version
|
||||||
//! without test files passes.
|
//! without test files passes.
|
||||||
|
|
||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
@@ -90,7 +90,7 @@ pub async fn test(broker: &Broker, id: &str, version: u64, component: &Component
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl Candidate<'_> {
|
impl Candidate<'_> {
|
||||||
/// Start an instance on a fresh copy of the state, without HTTP.
|
/// Start an instance on a fresh copy of the state, without network.
|
||||||
fn start(&self) -> Result<(Instance, Scratch)> {
|
fn start(&self) -> Result<(Instance, Scratch)> {
|
||||||
static N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
|
static N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
|
||||||
let n = N.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
let n = N.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||||
|
|||||||
+6
-2
@@ -126,7 +126,8 @@ impl std::fmt::Display for Exit {
|
|||||||
|
|
||||||
pub struct InstanceConfig {
|
pub struct InstanceConfig {
|
||||||
pub mounts: Vec<Mount>,
|
pub mounts: Vec<Mount>,
|
||||||
pub allow_http: bool,
|
/// Outgoing HTTP, and TCP, UDP, and DNS through `wasi:sockets`.
|
||||||
|
pub network: bool,
|
||||||
pub stderr_log: PathBuf,
|
pub stderr_log: PathBuf,
|
||||||
pub env: Vec<(String, String)>,
|
pub env: Vec<(String, String)>,
|
||||||
}
|
}
|
||||||
@@ -277,6 +278,9 @@ impl Runtime {
|
|||||||
.with_context(|| format!("open {}", cfg.stderr_log.display()))?;
|
.with_context(|| format!("open {}", cfg.stderr_log.display()))?;
|
||||||
|
|
||||||
let mut b = WasiCtx::builder();
|
let mut b = WasiCtx::builder();
|
||||||
|
if cfg.network {
|
||||||
|
b.inherit_network().allow_tcp(true).allow_udp(true).allow_ip_name_lookup(true);
|
||||||
|
}
|
||||||
b.stdin(AsyncStdinStream::new(guest_in));
|
b.stdin(AsyncStdinStream::new(guest_in));
|
||||||
b.stdout(AsyncStdoutStream::new(8192, guest_out));
|
b.stdout(AsyncStdoutStream::new(8192, guest_out));
|
||||||
b.stderr(AsyncStdoutStream::new(8192, tokio::fs::File::from_std(stderr)));
|
b.stderr(AsyncStdoutStream::new(8192, tokio::fs::File::from_std(stderr)));
|
||||||
@@ -293,7 +297,7 @@ impl Runtime {
|
|||||||
wasi: WasiPart { ctx: b.build(), table: ResourceTable::new() },
|
wasi: WasiPart { ctx: b.build(), table: ResourceTable::new() },
|
||||||
fsx: FsExt::new(cfg.mounts),
|
fsx: FsExt::new(cfg.mounts),
|
||||||
http: WasiHttpCtx::new(),
|
http: WasiHttpCtx::new(),
|
||||||
hooks: HttpHooks { allow: cfg.allow_http },
|
hooks: HttpHooks { allow: cfg.network },
|
||||||
};
|
};
|
||||||
|
|
||||||
let (line_tx, line_rx) = mpsc::channel(1024);
|
let (line_tx, line_rx) = mpsc::channel(1024);
|
||||||
|
|||||||
@@ -16,6 +16,15 @@ fn reply(out: &mut impl Write, message: &Value, body: Value) {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Connect to `address`, send one line, and return the line that comes back.
|
||||||
|
fn tcp_line(address: &str, line: &str) -> io::Result<String> {
|
||||||
|
let mut stream = std::net::TcpStream::connect(address)?;
|
||||||
|
writeln!(stream, "{line}")?;
|
||||||
|
let mut answer = String::new();
|
||||||
|
io::BufReader::new(stream).read_line(&mut answer)?;
|
||||||
|
Ok(answer.trim_end().to_string())
|
||||||
|
}
|
||||||
|
|
||||||
fn io_error(error: io::Error) -> Value {
|
fn io_error(error: io::Error) -> Value {
|
||||||
json!({"type": "error", "error": error.to_string()})
|
json!({"type": "error", "error": error.to_string()})
|
||||||
}
|
}
|
||||||
@@ -131,6 +140,10 @@ fn main() {
|
|||||||
json!({"type": "armed"})
|
json!({"type": "armed"})
|
||||||
}
|
}
|
||||||
Some("ignore-stdin") => json!({"type": "ignoring"}),
|
Some("ignore-stdin") => json!({"type": "ignoring"}),
|
||||||
|
Some("tcp") => match tcp_line(body["address"].as_str().unwrap_or_default(), text) {
|
||||||
|
Ok(line) => json!({"type": "tcp", "line": line}),
|
||||||
|
Err(error) => io_error(error),
|
||||||
|
},
|
||||||
Some("read") => match std::fs::read_to_string(&path) {
|
Some("read") => match std::fs::read_to_string(&path) {
|
||||||
Ok(text) => json!({"type": "content", "text": text}),
|
Ok(text) => json!({"type": "content", "text": text}),
|
||||||
Err(error) => io_error(error),
|
Err(error) => io_error(error),
|
||||||
|
|||||||
@@ -831,3 +831,34 @@ async fn candidates_receive_version_and_preexisting_spawn_state() {
|
|||||||
assert_eq!(successful, 2);
|
assert_eq!(successful, 2);
|
||||||
host.stop().await;
|
host.stop().await;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn running_actors_have_sockets_and_candidates_do_not() {
|
||||||
|
// A TCP server on the host that answers each line with "echo: <line>".
|
||||||
|
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||||
|
let address = listener.local_addr().unwrap().to_string();
|
||||||
|
tokio::spawn(async move {
|
||||||
|
while let Ok((stream, _)) = listener.accept().await {
|
||||||
|
tokio::spawn(async move {
|
||||||
|
use tokio::io::{AsyncBufReadExt, AsyncWriteExt};
|
||||||
|
let (read, mut write) = stream.into_split();
|
||||||
|
let mut lines = tokio::io::BufReader::new(read).lines();
|
||||||
|
while let Ok(Some(line)) = lines.next_line().await {
|
||||||
|
let _ = write.write_all(format!("echo: {line}\n").as_bytes()).await;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
let world = create_world("sockets");
|
||||||
|
let wasm = toolbox_wasm().clone();
|
||||||
|
let (host, mut mailbox) = start(&world).await;
|
||||||
|
let tcp = json!({"type": "tcp", "address": address, "text": "hi"});
|
||||||
|
// The candidate test expects the connection to fail.
|
||||||
|
let offline = source_with_test(&world, "offline", json!({"steps": [{"send": tcp, "expect": {"type": "error"}}]}));
|
||||||
|
assert_eq!(spawn_actor(&host, &mut mailbox, "net", &wasm, &offline).await.body["type"], "spawned");
|
||||||
|
|
||||||
|
let answer = ask(&host, &mut mailbox, "net", tcp).await;
|
||||||
|
assert_eq!(answer.body, json!({"type": "tcp", "line": "echo: hi"}));
|
||||||
|
host.stop().await;
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user