# threepm-io Application metrics, events and custom actions for a process started by [3pm](https://github.com/Unitech/3pm) — pm2's `io.metric` / `io.counter` / `io.notifyError` / `io.action` / `io.stream` / `io.open`, for 3pm, in Rust. Two dependencies (`serde`, `serde_json`), std threads, no runtime: it works the same in a sync program, under tokio, or in an audio callback's host. ## Install The crate is served from the 3pm download site, not from crates.io. Declare the registry once, in `.cargo/config.toml` (beside `Cargo.toml`, or in `$CARGO_HOME`): ```toml [registries.threepm] index = "sparse+https://3pm.edgecr4ft.net/sdk/cargo/" ``` then the dependency: ```toml [dependencies] threepm-io = { version = "0.64.0", registry = "threepm" } ``` `cargo add threepm-io --registry threepm` writes that line for you. The registry is public and needs no token. Actions take and return `serde_json::Value`: the crate re-exports it (`threepm_io::serde_json`), or add `serde_json = "1"` beside it to write `serde_json::json!` as the examples do; metrics and events need nothing else. Every version, its `.crate` file and its sha256: . In this repository a path dependency works too: `threepm-io = { path = "3pm/packages/rust" }`. ## Quickstart ```rust use threepm_io::{io, Event, Init, Level, Metric}; io().init(Init::new().source("worker").on_shutdown(|| { // Asked to stop: signal your own loop, do not block here. })); let requests = io().counter("http.requests", Metric::new().label("method", "GET")); let latency = io().histogram("http.latency", Metric::new().unit("ms")); let temp = io().gauge( "room.temp", Metric::new().unit("Cel").description("studio temperature"), ); requests.inc(1.0); latency.observe(12.5); temp.set(21.5); io().event( "deploy", Event::new().level(Level::Warn).attr("version", "1.2"), ); ``` Then `3pm show ` (section "Custom metrics") and `3pm events -f app`. Outside 3pm (no `THREEPM_CTL_SOCK` in the environment) every call is a no-op: no socket, no thread, no timer. Nothing here blocks the caller, returns a `Result` on telemetry, or panics — a metric that cannot exist (an invalid name, a reserved label, a name already taken by another kind) comes back as an inert handle, `enabled() == false`, plus one line on stderr under `THREEPM_IO_DEBUG=1`. `io()` is the process's instance, one control connection under one `source`. `Io::new()` makes another, for a test or a second source. ## Metrics ```rust let ok = requests.labels(&[("status", "200")]); ok.inc(1.0); // keep the child, do not derive it per request io().metric("queue.depth", Metric::new(), move || { read.load(std::sync::atomic::Ordering::Relaxed) as f64 }); io().meter("req.sec", Metric::new()).mark(1.0); io().histogram("db.query", Metric::new().unit("ms").scale(3)) .observe(0.8); ``` | Call | What | |---|---| | `gauge(name, Metric)` → `.set(v)` | The last value | | `counter(name, Metric)` → `.inc(n)` | Cumulative since process start; never decremented, rates computed at read time. Rust has no default arguments: `inc(1.0)`, not `inc()` | | `meter(name, Metric)` → `.mark(n)` | pm2's meter — the same counter under the name a rate is read by | | `histogram(name, Metric)` → `.observe(v)` | Base-2 exponential buckets, `scale` 0..=4 (default 1). A value that would open a 129th bucket lowers the scale by one, merging pairwise: `count` and `sum` stay exact | | `metric(name, Metric, f)` | A gauge read by calling `f` **on the flush thread**, once per `flush_ms`. `false` when the name could not be taken | | `.labels(&[(k, v)])` | A child series with those labels added. Same name + labels = the same series | `Metric::new().unit("ms").description("…").label(k, v).scale(2)` carries the options; only the first declaration of a name is kept. `Series` is `Clone + Send + Sync` — create the handles at startup and keep them: `.labels()` takes the registry lock and validates, which is fine per request and wasteful per sample. Non-finite values (NaN, infinities) are ignored, as is a negative `inc`. Names are `^[a-z][a-z0-9_.]{0,63}$`, the unit is **not** in the name (`http.latency` + `unit: "ms"`). At most 3 label keys (`^[a-z][a-z0-9_]{0,31}$`, values ≤ 64 bytes); the keys the fleet sets itself (`app`, `instance`, `src`, `node`, `overflow`, …) are refused. Past 50 label sets a metric folds into one `overflow="true"` series, whose `.labels()` returns the fold itself. Full model, bounds and the other languages: `docs/app-metrics.md` in the 3pm repository. ### Families, removal, usage — for series made from what a process finds A process that creates series from what it discovers (a monitoring module: one series per service found) needs three more things: ```rust use threepm_io::{io, Metric}; // A family: the name, unit and description — and no series. `gauge()` // reports its unlabelled series from the first flush, even when only // its children are ever set; a family never does. let memory = io().gauge_family("redis.memory", Metric::new().unit("By")); let used = memory.labels(&[("target", "redis@127.0.0.1:6379"), ("dim", "used")]); used.set(1_048_576.0); // The service is gone: stop reporting the series. The daemon drops it // after its TTL (two minutes); asking for the same labels again starts // a fresh one. `memory.remove_all()` removes every child. used.remove(); // Budget against `threepm_io::limits` before creating more. let u = io().usage(); // names, series, rejected, folded, removed ``` | Call | What | |---|---| | `gauge_family` / `counter_family` / `histogram_family(name, Metric)` → `Family` | The declaration alone; `.labels(&[(k, v)])` makes each child (the declaration's labels included) | | `Series::remove()` | Stop sending it; the handle and its clones become inert until the labels are asked for again | | `Family::remove_all()` | Every child; the name stays declared (it counts against the 200) | | `Io::usage()` → `Usage` | Names and series held, declarations refused, label sets folded, series removed | Under a daemon that announces `sticky_meta` in `THREEPM_CTL_CAPS`, a series' unit and description travel once per connection (again after a reconnect) instead of in every batch. ## Events and errors ```rust io().event( "midi.port_lost", Event::new() .level(Level::Warn) .attr("port", "ES-9") .attr("retries", 3) .fp("midi.port_lost"), ); if let Err(e) = std::fs::read("/does/not/exist") { io().notify_error(&e, Event::new().attr("where", "worker")); } ``` An event is sent at once (and queued, 64 of them, while the connection comes up). `Level` is `Debug` / `Info` / `Warn` / `Error` / `Fatal`; `attr` takes a string, a bool or any integer or float, at most 16 keys, strings cut to 256 bytes; `fp` (≤ 64 bytes) is the fingerprint the daemon groups and rate-limits by, derived from the event when you name none. `notify_error(&err, Event)` takes any `E: std::error::Error + ?Sized` and sends an `error`-level event named `error` carrying: - `exc.type` — `std::any::type_name::()`, or `Event::name("…")` when you want something else (a `Box` reports the box); - `exc.message` — the error's `Display`; - `exc.stacktrace` — its `source()` chain, then a backtrace **of the `notify_error` call site**, not of where the error was made, and only when `RUST_BACKTRACE` or `RUST_LIB_BACKTRACE` enables one. An empty stacktrace is left off the wire. `Event::exception(type, message)` and `Event::stacktrace(…)` attach an exception by hand to an event of any name. ## Actions An action is a function of your process that `3pm trigger` and the fleet backend can call (pm2's `io.action`). ```rust // A JSON result: at most 512 KiB, at most 64 levels deep. io().action( "cache.flush", Action::new() .description("drop the cache") .timeout_ms(5_000), |params: Value, _ctx: &ActionCtx| -> Result { let prefix = params["prefix"].as_str().unwrap_or(""); Ok(json!({ "flushed": prefix })) }, ); // A failure: anything that prints. `Err` is the call's error message. io().action("reindex", Action::new(), |_: Value, _: &ActionCtx| { std::fs::read_to_string("/etc/hostname") .map(|host| json!({ "host": host.trim() })) .map_err(|e| format!("reindex failed: {e}")) }); ``` The handler is `Fn(Value, &ActionCtx) -> Result + Send + Sync` with `E: Display` — `String`, `io::Error`, `Box`, `anyhow::Error`, your own type. It runs on its own thread and may block for as long as its timeout allows. `Ok(v)` is the call's `data` (`Ok(Value::Null)` answers success with no data), `Err(e)` its error message (cut to 1024 bytes). A handler that panics answers `the handler panicked: …` and the process lives on. **A file, for a result too big for one line.** The daemon prepares a path; write `.tmp` and rename it into place with `commit_file`: ```rust io().action( "dump", Action::new().output_max_bytes(16 << 20), |_: Value, ctx: &ActionCtx| -> Result { let tmp = ctx.tmp_file().ok_or("the call named no file")?; std::fs::write(&tmp, b"{}").map_err(|e| e.to_string())?; Ok(json!({ "bytes": ctx.commit_file("json").map_err(|e| e.to_string())? })) }, ); ``` `commit_file(ext)` takes the extension (`^[a-z0-9]{1,16}$`, never `tmp`) and returns the size. It refuses — deleting the `.tmp` — in the call's last tenth (at most the last second), because the daemon has given up and swept by then, and past the `output_max_bytes` the action declared. `output_max_bytes` outside 1..=256 MiB leaves the action **unregistered**: the daemon refuses a source's whole list over one bad declaration, so the SDK refuses the one action instead. **Local-only, and concurrency.** ```rust let reset = io().action( "reset", Action::new().remote(false).concurrent(4), |_: Value, _: &ActionCtx| Ok::(json!("done")), ); if !reset.enabled() { eprintln!("reset was refused: run with THREEPM_IO_DEBUG=1 to see why"); } reset.unregister(); // explicit; dropping the handle withdraws nothing ``` `remote(false)` keeps the action out of the fleet backend's reach (`3pm trigger` still calls it). `concurrent(n)`, 1..=16, is how many calls of that action may run at once — past it a call is answered `busy`, and a slot is held at most the call's timeout, so a hung handler frees it when the daemon gives up. `Registration::unregister()` is explicit on purpose: a silent `Drop` would withdraw an action by accident. **Bytes the caller sent.** ```rust io().action( "import", Action::new().input(8 << 20, &["json", "csv"]), |_: Value, ctx: &ActionCtx| -> Result { let input = ctx.input.as_ref().ok_or("this call carried no bytes")?; let body = std::fs::read(&input.path).map_err(|e| e.to_string())?; Ok(json!({ "read": body.len(), "ext": input.ext })) }, ); ``` `input(max_bytes, ext)` is the declaration — at most 8 GiB, at most 16 extensions. `ctx.input` is `ActionInput { path, bytes, ext, sha256, live }`: the bytes are already on this host, staged by the daemon. **Read them, never delete them.** A call whose input is a *blob* (bytes the fleet backend holds, fetched over `data.sock`) is refused at once with a message saying so — that is not implemented yet, see below. Names are `^[A-Za-z0-9_.:-]{1,64}$`, 32 per source; `3pm:` is the SDK's prefix and a name under it is refused (an inert `Registration`, never a panic). Registering a name twice replaces the handler and keeps its place; the older `Registration` then withdraws nothing. The whole list is declared right after every `hello` and again whenever it changes, and re-sent every flush until a write of it succeeded. `3pm trigger ` lists what is declared. ## Streams A stream is a call with two framed legs instead of one result: the caller and the callee each `send` frames the other `recv`s, until both ended. Frames ride the agent's `data.sock` — agent to agent when the two nodes are LAN neighbours, through the fleet backend otherwise — so both sides must run on **linked** nodes (`3pm link`); a session to this very host goes through the fleet too. **The callee** declares a stream action; its handler runs once per session, on its own thread, for as long as the session lasts: ```rust io().stream( "echo", Action::new() .description("send every frame back") .concurrent(4) .idle_ms(60_000), |s: &Session, _params: Value| -> Result { while let Some(frame) = s.recv().map_err(|e| e.to_string())? { // Forwarded as it came: payload, flags and `ts_us`, so the // caller measures the round trip from its own stamp. s.send_frame(frame); } Ok(Value::Null) }, ); io().serve(); ``` Returning closes the session; `Ok(Value::Null)` answers its stats. **The caller** opens a session on a node (`Some("3pmn_…")`, a unique hostname, or `None` / `"self"` for this host): ```rust let s = io().open(node, "echo", "echo", json!({}), Open::new())?; s.send(b"hello"); // no flag, stamped now s.send_frame(Frame::new(b"a key frame".to_vec()).with(FLAG_KEY)); s.end(); // half-close: our END, we keep reading while let Some(frame) = s.recv()? { println!("{} bytes back", frame.payload.len()); } let reply = s.result(Duration::from_secs(5)); // Some(Ok(data)) once answered s.close(); ``` `open` blocks until the daemon's ticket (≤ 5 s, `daemon_too_old` past it). A refusal before the ticket is its `Err`; one after it (`busy`, a target whose `callers` say no) fails the session — `recv` returns that error, `send` returns `false`, `result` says so. | `Session` | | |---|---| | `send(&[u8])`, `send_frame(Frame)`, `send_shared(&Arc<[u8]>)` | Queue one frame, never block; `false` once this side ended or the session failed, or for a payload over 1 MiB. `send` copies the payload once, `send_frame` moves it, `send_shared` shares it — a frame broadcast to N sessions is held once | | `recv()` | The next frame, blocking; `Ok(None)` at the other side's `END`, after `close`, and on a callee whose caller sends nothing; `Err(CallError)` when a leg broke, a minute passed without a byte, or the call failed | | `end()` | Half-close: our `END`, then the agent's answer (≤ 3 s); keep reading | | `close()` | `end()`, then stop reading: a `recv` blocked in another thread returns `Ok(None)`; the session's slot comes back, and the call stops counting against the 16 in flight (its reply, if it has not come, still has the reply grace, then `lost`) | | `stats()` | `sent`, `dropped`, `bytes_out`, `received`, `bytes_in`, `missing` (holes in the other side's `seq`, every abandon point on the path), `discont` | | `failure()`, `call_id()` | What killed the session; the call's id | | `result(wait)` | Caller only: the callee's answer — its handler's data, or its stats when it returned `Null` | `Session` is `Send + Sync` and `Clone`: one thread `recv`s while another `send`s. Dropping the last clone ends the writing leg without waiting. **Lossy by design, low latency first.** A frame is written as soon as it is sent — one HTTP chunk per frame, on its own writer thread, so `send` never waits; what is already queued when the writer comes round goes out in one vectored write (up to 64 frames, 64 KiB), but nothing is held back to make a batch. While that socket is blocked (the agent or the link applies back-pressure) frames queue: a frame that waited past **100 ms** is dropped on its own age (never the newest one), and past **8 MiB** the oldest frame that is not a `KEY` goes first, then the oldest `KEY` or superseded `INIT`, until the queue fits. Only the newest `INIT` is kept whatever its age, as the agent's sink keeps it: drops happen at the source and only while the socket is blocked, never on a socket that keeps up. Mark the frames a reader can start from `FLAG_KEY` and the decoding parameters `FLAG_INIT`; an empty payload is a fine keepalive. A live leg keeps nothing for a reader that has not joined yet: frames sent right after `open`, before the callee's handler first `recv`s, can be lost on the way (they show up in the callee's `missing`). A caller that needs its first frames delivered sends a hello until the first frame comes back. **Every leg must move at least once a minute.** A reading leg gives up after 60 s without a byte — in either direction, and whatever `idle_ms` the action declares (that one is the daemon's guard, this one the backend's). A session that may stay quiet sends an empty frame every 15 s or so. `Open::window_ms` / `max_bytes` move the window, `settle_ms` the `end` wait, `reply_grace_ms` how long the reply has once a leg broke under the caller (10 s, then `lost`). Each side holds two sockets and two threads per session (the writer, and a watcher of the agent's answer, which turns an early `401`/`422`/`503` into the session's failure at once). The idle period (`idle_ms`, the declaration's `timeout_ms`) is enforced by the daemon; a session never outlives 24 h. `io().node()` reads this host's node from the agent's `link.toml` (`None` when it is not linked). **A daemon reconnect does not end a session.** The legs live on the data socket, not on the control connection: when that connection drops, a call that already has its ticket gets the reply grace (`reply_grace_ms`, 10 s) and is `lost` only if no reply came by then — the session keeps moving meanwhile. A call still waiting for its ticket fails at once (`lost`). The Node SDK fails both at once. Errors are `CallError { code, message, call_id }`, printed `code: message`; `code` is one of `CallError::{TIMEOUT, REFUSED, TOO_LARGE, BUSY, GONE, FAILED, LOST, NODE_OFFLINE, NODE_UNKNOWN, AMBIGUOUS, FORBIDDEN, AGENT_TOO_OLD, DAEMON_TOO_OLD}` or a newer one. A status of the agent maps as the Node SDK's: `401`/`403` forbidden, `404`/`409` gone, `408` timeout, `413`/`422` too_large, `429` busy, `503` node_offline. `examples/stream_echo.rs` is both sides. ## Lifecycle ```rust if io().ready(Duration::from_millis(500)) { io().event("boot", Event::new()); } io().flush(Duration::from_millis(500)); io().shutdown(Duration::from_millis(500)); ``` | Call | What | |---|---| | `init(Init)` | Connect in the background. Idempotent, and usable again after `shutdown`. Without `THREEPM_CTL_SOCK` the instance stays disabled | | `enabled()` | Is a connection *configured* (not: up)? | | `source()`, `stats()` | The name this instance said hello with; `{sent, dropped, connects, closes, refusals}` | | `ready(Duration)` / `ready_default()` | Wait for the connection — for a program that wants to know before it reports. `ready_default` waits `connect_wait_ms` (5 s) | | `flush(Duration)` | Send the whole registry now. With a non-zero wait, wait for the connection, then for the bytes to leave the process | | `shutdown(Duration)` | `flush`, then close. `serve()` returns | | `serve()` | Block until the daemon asks the process to stop | **A Rust process ends when `main` returns.** No thread of this SDK holds it up (Node's `keepAlive` has no equivalent), so a worker whose only job is to answer actions ends its `main` with `io().serve()`: it blocks until the daemon's `shutdown`, until `Io::shutdown`, or until the transport gives up for good. On an instance that was never `init`ed it returns at once. **There is no `atexit`.** A short-lived program must call `io().flush(Duration::from_millis(500))` — or `shutdown` — before returning, or its last batch never leaves. A long-running one needs neither: the registry goes out every `flush_ms`. `Init` carries `source`, `flush_ms` (1 s), `connect_wait_ms` (5 s), `on_shutdown` and `env` (read the identity from a map instead of the process environment), plus the test knobs the three other SDKs spell the same way: `reconnect_min_ms`, `reconnect_max_ms`, `max_buffer_bytes`, `max_refusals`, `refusal_window_ms`. `on_shutdown` runs on the reader thread — signal, do not block. A source names one process within the app (`^[a-z0-9][a-z0-9_-]{0,31}$`). Named none, the process 3pm spawned takes `main`, and a process *it* started takes `pid-`; two that both claim `main` settle it — the refused one falls back once, and that refusal never counts toward giving up. ## Bounds Everything the SDK holds itself to, so that nothing it writes is ever refused. `threepm_io::limits` exposes each one as a constant. | | | |---|---| | Metric names | 200 per instance, `^[a-z][a-z0-9_.]{0,63}$` | | Labels | 3 keys per metric, values ≤ 64 bytes, 50 label sets then an `overflow="true"` fold | | Histogram | 128 non-empty buckets, `scale` 0..=4 | | Batches | 256 entries and 32 KiB per `metrics` line | | Events | 16 attributes, strings ≤ 256 bytes, `exc.type` ≤ 128, `exc.message` ≤ 1 KiB, `exc.stacktrace` ≤ 8 KiB, `fp` ≤ 64 | | Actions | 32 per source, `description` ≤ 128 bytes, `timeout_ms` ≤ 10 min (30 s by default), `concurrent` ≤ 16, result ≤ 512 KiB and ≤ 64 levels | | Streams | 16 sessions per instance (both directions), 16 calls in flight per connection (a session holds one until its reply), frames ≤ 1 MiB, idle ≤ 24 h, a reading leg gives up after 60 s without a byte | | Telemetry buffer | 64 KiB; past it a `metrics` line is dropped and the next one carries the same totals (counters and buckets are cumulative) | | Control buffer | 5 × 1 MiB of its own: the actions declaration and the replies can never be crowded out by metrics | | Reconnection | 1 s → 30 s, doubling; the instance gives up for good after 5 refused `hello`s (the process outlived its incarnation) | Longer strings are cut on a UTF-8 character boundary, not refused. A value that cannot be fixed (a bad name, a reserved label) makes the handle inert instead. ## Differences from the Node SDK - **Not implemented yet**: `io.call` (calling another node's action) and **blob** action inputs and outputs (a file that crosses the fleet). A call carrying a blob input is refused with a message that says so; an output ticket is ignored, and the handler's committed file is what the caller gets. Streams are there (`io().stream` / `io().open`). - **Streams are `send` / `recv`**, not Node's `session.input` writable and `frames()` iterator: no `pipe`, no `mjpeg`, and `stats()` carries no latency percentiles. A session's slot comes back at `close` (or when it drops, or fails), not when the reply arrives. - **A control reconnect gives an open session its reply grace** rather than failing it at once (see Streams). - **No `3pm:profile`.** `3pm profile` returns nothing for a Rust process; `3pm:` names are refused rather than reserved for a built-in. - **No built-in runtime metrics.** Node's `builtins: true` (event-loop delay, heap, GC, handles) has no Rust counterpart; declare what you want with `metric(…)`. - **`Event::name` names the exception type**, not the event. Node uses `opts.name` as the event name and takes `exc.type` from the JS error class; Rust has no error class, and `event(name, …)` already takes the name as its first argument. So `notify_error` always sends the event `error`, and `Event::name` overrides `exc.type`. For a custom event name carrying an exception: `io().event("my.failure", Event::new().exception(kind, message))`. - **`inc(n)` and `mark(n)` take their step**, where Node defaults to 1. - **`metric` callbacks run on the flush thread**, not on an event loop. Read a plain value there; do not block, and do not take a lock the rest of your program holds while writing metrics. - **An `action` line read past its deadline is not run and not answered.** The daemon has already answered `timeout` and swept the call's file, so a handler would work on a path that is gone and a reply would be discarded. Node runs it anyway. - **`action_reply` carries no `code`.** `too_large` and `busy` travel as plain failures whose message says why; the wire has no such field, and the daemon ignores the one Node sends. - **A `3pm:` name is an inert `Registration`, not a throw**, and no telemetry call returns a `Result` — an SDK must not kill a worker over a name. - **No `worker-*` source**: a Rust thread shares its process's connection. ## Debugging `THREEPM_IO_DEBUG=1` in the app's environment prints one `[3pm io]` line on stderr for every refusal — an invalid name, a reserved label, a cut description, an action the table would not take, a source that fell back. It is read once, at `Io::new`. `io().stats()` gives the counters the transport keeps. ## Tests `cargo test` in this directory: unit tests against a fake `ctl.sock`, integration tests decoding every line the SDK writes with `threepm-protocol` (the crate the daemon itself uses), the frozen `limits.json` and `exphist-vectors.json` gates, and proptests over the histogram. `cargo build --examples` builds the snippets above (`examples/quickstart.rs`, `examples/worker.rs`) and `examples/e2e_client.rs`, which `crates/threepm-cli/tests/e2e_io.rs` and `e2e_actions.rs` run under a real daemon.