carvers/anwesen
5
0
Fork 0
Code Releases 4 Activity Actions

Compare commits

base: carvers:main
Branches Tags
carvers:main
carvers:v0.4.0
carvers:v0.3.0
carvers:v0.2.0
carvers:v0.1.0
..
compare: carvers:v0.2.0
Branches Tags
carvers:main
carvers:v0.4.0
carvers:v0.3.0
carvers:v0.2.0
carvers:v0.1.0

No commits in common. "main" and "v0.2.0" have entirely different histories.
main ... v0.2.0

16 changed files with 464 additions and 1114 deletions
Download patch file Download diff file Expand all files Collapse all files

2
Cargo.lock generated
View file

@ -72,7 +72,7 @@ dependencies = [
[[package]] [[package]]
name = "anwesen" name = "anwesen"
version = "0.4.0" version = "0.2.0"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"axum", "axum",

2
Cargo.toml
View file

@ -1,7 +1,7 @@
[package] [package]
name = "anwesen" name = "anwesen"
description = "Read-only HTTP daemon over a markdown vault, querying YAML frontmatter." description = "Read-only HTTP daemon over a markdown vault, querying YAML frontmatter."
version = "0.4.0" version = "0.2.0"
edition = "2024" edition = "2024"
rust-version = "1.95" rust-version = "1.95"
license = "BSD-3-Clause" license = "BSD-3-Clause"

97
README.md
View file

@ -22,7 +22,7 @@ Anwesen answers three kinds of question over HTTP:
The frontmatter index is built once at startup and kept current by watching the vault directory. The index lives in memory; a restart rebuilds it, and there is nothing on disk to corrupt or migrate. The frontmatter index is built once at startup and kept current by watching the vault directory. The index lives in memory; a restart rebuilds it, and there is nothing on disk to corrupt or migrate.
The same query-and-merge engine also runs offline, with no server. `anwesen merge` walks a directory, evaluates a query, and writes the merged markdown to stdout; `anwesen query` writes the same JSON document `GET /query` returns (see [Local generation](#local-generation)). The same query-and-merge engine also runs offline, with no server: `anwesen merge` walks a directory, evaluates a query, and writes the merged markdown to stdout (see [Local generation](#local-generation)).
## Quick start ## Quick start
@ -52,20 +52,14 @@ Build one file out of many notes, without starting the daemon:
anwesen merge --vault /path/to/vault --query 'tags=adr&__anw-order=title' > ADRs.md anwesen merge --vault /path/to/vault --query 'tags=adr&__anw-order=title' > ADRs.md
``` ```
Ask which notes match, and what their frontmatter holds, without starting the daemon:
```
anwesen query --vault /path/to/vault --query 'tags=adr' | jq -r '.results[].path'
```
## CLI ## CLI
``` ```
anwesen serve --vault <path> [--bind <addr:port>] [--log-level <level>] anwesen serve --vault <path> [--bind <addr:port>] [--log-level <level>]
[--otlp-slow-request-ms <n>] [--uptrace-dsn <dsn> | --otlp-endpoint <url>]
[--otlp-header <key=value>]... [--otlp-slow-request-ms <n>]
anwesen doctor --vault <path> anwesen doctor --vault <path>
anwesen merge --vault <path> --query <query-string> anwesen merge --vault <path> --query <query-string>
anwesen query --vault <path> --query <query-string>
anwesen version anwesen version
``` ```
@ -74,64 +68,25 @@ anwesen version
| `--vault <path>` | `ANWESEN_VAULT` | _required_ | Path to the vault root. | | `--vault <path>` | `ANWESEN_VAULT` | _required_ | Path to the vault root. |
| `--bind <addr:port>` | `ANWESEN_BIND` | `127.0.0.1:8080` | Listen address for `serve`. | | `--bind <addr:port>` | `ANWESEN_BIND` | `127.0.0.1:8080` | Listen address for `serve`. |
| `--log-level <level>` | `ANWESEN_LOG_LEVEL` | `info` | `error`, `warn`, `info`, `debug`, or `trace`. | | `--log-level <level>` | `ANWESEN_LOG_LEVEL` | `info` | `error`, `warn`, `info`, `debug`, or `trace`. |
| `--query <query-string>` | `ANWESEN_QUERY` | empty (match all) | A `/query` query string: frontmatter predicates plus `__anw-` controls. `merge` and `query` only. | | `--query <query-string>` | -- | _required for `merge`_ | A `/query` query string: frontmatter predicates plus `__anw-` controls. |
| `--uptrace-dsn <dsn>` | `ANWESEN_UPTRACE_DSN` | unset | uptrace DSN (`https://<token>@api.uptrace.dev`). Excludes `--otlp-endpoint`. |
| `--otlp-endpoint <url>` | `ANWESEN_OTLP_ENDPOINT` | unset | OTLP/HTTP base URL. Excludes `--uptrace-dsn`. |
| `--otlp-header <key=value>` | `ANWESEN_OTLP_HEADERS` | none | Extra export header, repeatable. The env var takes a comma-separated list. |
| `--otlp-slow-request-ms` | `ANWESEN_OTLP_SLOW_REQUEST_MS` | `500` | Requests at or over this duration, or answering 5xx, also export a span. | | `--otlp-slow-request-ms` | `ANWESEN_OTLP_SLOW_REQUEST_MS` | `500` | Requests at or over this duration, or answering 5xx, also export a span. |
Every flag has a matching `ANWESEN_<UPPER>` environment variable. CLI flags win Every flag has a matching `ANWESEN_<UPPER>` environment variable, except
over env vars. `--otlp-header`, whose env var is the plural `ANWESEN_OTLP_HEADERS` because it
takes a list. CLI flags win over env vars.
`--bind` and `--otlp-slow-request-ms` apply to `serve` only. Where telemetry is Telemetry is off unless `--uptrace-dsn` or `--otlp-endpoint` is set. With
exported is configured entirely through the standard `OTEL_` variables below. neither, nothing is exported and no exporter is built. The four telemetry
flags apply to `serve` only.
- **`serve`** -- run the daemon: walk the vault, build the index, watch for changes, serve the API. - **`serve`** -- run the daemon: walk the vault, build the index, watch for changes, serve the API.
- **`doctor`** -- walk the vault once and report what would stop clean ingestion: unreadable files, unparseable YAML, path collisions on the HTTP surface, and frontmatter type drift (the same key carrying incompatible types across notes). Read-only; non-zero exit if any issue is found. - **`doctor`** -- walk the vault once and report what would stop clean ingestion: unreadable files, unparseable YAML, path collisions on the HTTP surface, and frontmatter type drift (the same key carrying incompatible types across notes). Read-only; non-zero exit if any issue is found.
- **`merge`** -- one-shot local generation: walk the vault, evaluate `--query`, and write the merged markdown document to stdout. No server, no HTTP. See [Local generation](#local-generation). - **`merge`** -- one-shot local generation: walk the vault, evaluate `--query`, and write the merged markdown document to stdout. No server, no HTTP. See [Local generation](#local-generation).
- **`query`** -- the same one-shot walk, writing the JSON document `GET /query` returns: which notes match, and what their frontmatter, `last_modified`, `etag` and `size` hold. No server, no HTTP. See [Local generation](#local-generation).
- **`version`** -- print version and exit. - **`version`** -- print version and exit.
## Telemetry
anwesen exports OTLP metrics for every request, and a span for requests at or
over `--otlp-slow-request-ms` or answering a 5xx. Export is configured through
the standard OpenTelemetry environment variables, which the SDK reads directly:
| Variable | Meaning |
| -------------------------------------- | ---------------------------------------------------------------- |
| `OTEL_EXPORTER_OTLP_ENDPOINT` | Base URL. The SDK appends `/v1/metrics` and `/v1/traces`. |
| `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` | Full metrics URL, used as given. Overrides the base for metrics. |
| `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` | Full traces URL, used as given. Overrides the base for traces. |
| `OTEL_EXPORTER_OTLP_HEADERS` | Export headers, `key=value` comma-separated. |
| `OTEL_EXPORTER_OTLP_PROTOCOL` | `http/protobuf`, the default and the only value this build speaks. |
| `OTEL_SERVICE_NAME`, `OTEL_RESOURCE_ATTRIBUTES` | Override the `anwesen` service identity. |
The per-signal `OTEL_EXPORTER_OTLP_METRICS_PROTOCOL` and
`OTEL_EXPORTER_OTLP_TRACES_PROTOCOL` are read the same way.
With none of the three endpoint variables set, telemetry is off: no exporter is
built and the request middleware is not installed.
Two settings fail at startup rather than export nowhere in silence:
- A query or a fragment on `OTEL_EXPORTER_OTLP_ENDPOINT`. The base URL must be
a base URL; the SDK appends the signal path after whatever it is given.
- A protocol other than `http/protobuf`. The binary ships the OTLP/HTTP
transport alone, so `grpc` would keep exporting over HTTP with nothing in the
log. Point the endpoint at the collector's HTTP port, not its gRPC one.
uptrace:
```
OTEL_EXPORTER_OTLP_ENDPOINT=https://api.uptrace.dev
OTEL_EXPORTER_OTLP_HEADERS=uptrace-dsn=https://TOKEN@api.uptrace.dev?grpc=4317
```
Paste the DSN from Project Settings into the header verbatim, tail and all --
it is a credential there, not an address. The endpoint is the host alone.
Removed in 0.3.0: `--uptrace-dsn`, `--otlp-endpoint`, `--otlp-header` and their
`ANWESEN_` variables. Passing any of them fails at startup with the `OTEL_`
replacement to use.
## HTTP API ## HTTP API
All endpoints are `GET` and return JSON unless noted. All endpoints are `GET` and return JSON unless noted.
@ -181,7 +136,7 @@ ISO-8601 dates and RFC 3339 datetimes are coerced to typed dates at read time, s
| `__anw-order=<key>[:asc\|:desc]` | path order | Order fragments (merge mode only). | | `__anw-order=<key>[:asc\|:desc]` | path order | Order fragments (merge mode only). |
| `__anw-kind=<key>` | off | Refuse a mixed merge unless every matched note shares one value for the key (merge mode only). | | `__anw-kind=<key>` | off | Refuse a mixed merge unless every matched note shares one value for the key (merge mode only). |
By default `/query` returns metadata only; fetch bodies with `/notes/<path>`. The same JSON document is available offline, without the daemon, via `anwesen query` (see [Local generation](#local-generation)). By default `/query` returns metadata only; fetch bodies with `/notes/<path>`.
#### Markdown-merge mode #### Markdown-merge mode
@ -209,11 +164,7 @@ Returns vault path, note count, last index/event timestamps, watcher state, an i
## Local generation ## Local generation
Two subcommands answer a query on the command line, with no server and no HTTP round-trip: `merge` writes the merged markdown document, `query` writes the JSON projection. Both take the same `--vault` and `--query` flags, both walk the vault once, and both run the engine the endpoint runs -- so the output matches what the daemon would have returned for the same vault and query. `anwesen merge` produces the markdown-merge document on the command line, with no server and no HTTP round-trip. It walks the vault, evaluates the query, and writes the merged document to stdout:
### `merge`
`anwesen merge` walks the vault, evaluates the query, and writes the merged document to stdout:
``` ```
anwesen merge --vault /path/to/vault --query 'tags=adr&__anw-order=title&__anw-kind=kind' anwesen merge --vault /path/to/vault --query 'tags=adr&__anw-order=title&__anw-kind=kind'
@ -223,24 +174,6 @@ The `--query` string is the exact `/query` grammar: frontmatter predicates plus
This is the materialization path: build a `CLAUDE.md`, a skill bundle, or any single file assembled from many notes, driven from a script or a one-off shell. This is the materialization path: build a `CLAUDE.md`, a skill bundle, or any single file assembled from many notes, driven from a script or a one-off shell.
### `query`
`anwesen query` answers the other half: which notes match, and what their frontmatter holds. It writes the same JSON document `GET /query` returns -- `results`, `total`, `truncated`, with `path`, `frontmatter`, `last_modified`, `etag` and `size` per row -- on one line, ready for `jq`:
```
anwesen query --vault /path/to/vault --query 'kind=PDR&__anw-limit=1'
```
```json
{"results":[{"path":"Projects/PDR-001-intro.md","frontmatter":{"kind":"PDR","num":1,"title":"PDR-001 Intro"},"last_modified":"2026-05-14T17:05:08Z","etag":"\"f657eab5...\"","size":57}],"total":2,"truncated":true}
```
`total` counts the full match set; `truncated` says the cap cut it.
Bodies are elided here as they are on the endpoint; `merge` is the way to get them offline. `__anw-order` and `__anw-kind` are merge-mode controls and do not affect this output. A malformed query or an unreadable vault exits non-zero with the reason on stderr; an empty match set is an empty `results` list and exit `0`.
A script can therefore move between the daemon and the CLI without a second parser: same shape, same field names, same timestamp dialect.
## Design notes ## Design notes
- **In place, read-only.** Anwesen reads the same directory Obsidian writes to and never writes back. The vault stays editable in Obsidian with no coordination, and there is no write API by design. - **In place, read-only.** Anwesen reads the same directory Obsidian writes to and never writes back. The vault stays editable in Obsidian with no coordination, and there is no write API by design.

13
src/app.rs
View file

@ -121,12 +121,6 @@ pub struct Anwesen {
/// Request-level telemetry handle ([ANW-37]). `None` disables export and /// Request-level telemetry handle ([ANW-37]). `None` disables export and
/// the request middleware entirely. /// the request middleware entirely.
pub telemetry: Option<Arc<Telemetry>>, pub telemetry: Option<Arc<Telemetry>>,
/// Set once the supervisor tree is up. Hydra's `Application::run` logs a
/// start failure and returns normally, so `serve` cannot tell a clean
/// shutdown from a tree that never came up. `main` reads this after `run`
/// and exits nonzero when it is still false, which is what lets systemd
/// retry ([ANW-45](https://crvrs.youtrack.cloud/issue/ANW-45)).
pub started: Arc<AtomicBool>,
} }
impl Anwesen { impl Anwesen {
@ -139,7 +133,6 @@ impl Anwesen {
store: NoteStore::new(), store: NoteStore::new(),
health: HealthState::new(), health: HealthState::new(),
telemetry, telemetry,
started: Arc::new(AtomicBool::new(false)),
} }
} }
} }
@ -194,12 +187,10 @@ impl Application for Anwesen {
.child_spec(), .child_spec(),
]; ];
let pid = Supervisor::with_children(children) Supervisor::with_children(children)
.strategy(SupervisionStrategy::OneForOne) .strategy(SupervisionStrategy::OneForOne)
.start_link(SupervisorOptions::new().name("anwesen_root")) .start_link(SupervisorOptions::new().name("anwesen_root"))
.await?; .await
self.started.store(true, Ordering::Release);
Ok(pid)
} }
} }

124
src/cli.rs
View file

@ -4,27 +4,18 @@
//! //!
//! ```text //! ```text
//! anwesen serve --vault <path> [--bind <addr:port>] [--log-level <level>] //! anwesen serve --vault <path> [--bind <addr:port>] [--log-level <level>]
//! [--otlp-slow-request-ms <n>] //! [--uptrace-dsn <dsn> | --otlp-endpoint <url>]
//! [--otlp-header <key=value>]... [--otlp-slow-request-ms <n>]
//! anwesen doctor --vault <path> [--log-level <level>] //! anwesen doctor --vault <path> [--log-level <level>]
//! anwesen merge --vault <path> [--query <string>] [--log-level <level>] //! anwesen merge --vault <path> [--query <string>] [--log-level <level>]
//! anwesen query --vault <path> [--query <string>] [--log-level <level>]
//! anwesen version //! anwesen version
//! ``` //! ```
//! //!
//! `merge` and `query` differ only in what they write -- the merged markdown //! `--bind` and the OTLP telemetry flags are `serve`-only ([ANW-37]);
//! document or the `/query` JSON projection ([ANW-43]) -- so they share one
//! [`VaultQueryArgs`] flag set and cannot drift.
//!
//! `--bind` and `--otlp-slow-request-ms` are `serve`-only ([ANW-37]);
//! `doctor` and `merge` do not bind a port; `version` takes no flags. Each //! `doctor` and `merge` do not bind a port; `version` takes no flags. Each
//! flag has a matching `ANWESEN_<UPPER>` environment variable and CLI wins //! flag has a matching `ANWESEN_<UPPER>` environment variable and CLI wins
//! over env per the manual. //! over env per the manual. With no `--otlp-endpoint`/`--uptrace-dsn`,
//! //! telemetry is off and the server behaves exactly as without these flags.
//! Telemetry export is configured through the standard `OTEL_EXPORTER_OTLP_*`
//! environment variables ([ANW-42](https://crvrs.youtrack.cloud/issue/ANW-42)).
//! With none of them set, telemetry is off and the server behaves as it did
//! before telemetry existed. The removed flags below are still parsed so an
//! upgrade fails loudly instead of dropping export config in silence.
use std::net::SocketAddr; use std::net::SocketAddr;
use std::path::PathBuf; use std::path::PathBuf;
@ -46,11 +37,7 @@ pub enum Command {
Doctor(DoctorArgs), Doctor(DoctorArgs),
/// Walk the vault once, evaluate the query, and write the merged markdown /// Walk the vault once, evaluate the query, and write the merged markdown
/// document to stdout. One-shot; no server. Read-only. /// document to stdout. One-shot; no server. Read-only.
Merge(VaultQueryArgs), Merge(MergeArgs),
/// Walk the vault once, evaluate the query, and write the same JSON
/// document `GET /query` returns to stdout. One-shot; no server.
/// Read-only.
Query(VaultQueryArgs),
/// Print the version and exit. /// Print the version and exit.
Version, Version,
} }
@ -69,22 +56,25 @@ pub struct ServeArgs {
#[arg(long, env = "ANWESEN_LOG_LEVEL", default_value = "info")] #[arg(long, env = "ANWESEN_LOG_LEVEL", default_value = "info")]
pub log_level: LogLevel, pub log_level: LogLevel,
/// Removed (ANW-42): use `OTEL_EXPORTER_OTLP_ENDPOINT` plus /// uptrace DSN shorthand (`https://<token>@api.uptrace.dev`), parsed
/// `OTEL_EXPORTER_OTLP_HEADERS=uptrace-dsn=<dsn>`. Still accepted so /// into the OTLP endpoint plus an `uptrace-dsn` header. Mutually
/// startup fails with that message rather than exporting nowhere. /// exclusive with --otlp-endpoint. When this and --otlp-endpoint are
#[arg(long, env = "ANWESEN_UPTRACE_DSN", hide = true)] /// both unset, telemetry is fully off (ANW-37).
#[arg(long, env = "ANWESEN_UPTRACE_DSN")]
pub uptrace_dsn: Option<String>, pub uptrace_dsn: Option<String>,
/// Removed (ANW-42): use `OTEL_EXPORTER_OTLP_ENDPOINT`. /// Generic OTLP/HTTP endpoint base URL for telemetry export. The
#[arg(long, env = "ANWESEN_OTLP_ENDPOINT", hide = true)] /// per-signal path (`/v1/metrics`, `/v1/traces`) is appended by the
/// exporter. Mutually exclusive with --uptrace-dsn.
#[arg(long, env = "ANWESEN_OTLP_ENDPOINT")]
pub otlp_endpoint: Option<String>, pub otlp_endpoint: Option<String>,
/// Removed (ANW-42): use `OTEL_EXPORTER_OTLP_HEADERS`. /// Extra OTLP export header as `key=value`, repeatable. On the env var
/// (`ANWESEN_OTLP_HEADERS`) pass a comma-separated `key=value` list.
#[arg( #[arg(
long = "otlp-header", long = "otlp-header",
env = "ANWESEN_OTLP_HEADERS", env = "ANWESEN_OTLP_HEADERS",
value_delimiter = ',', value_delimiter = ','
hide = true
)] )]
pub otlp_headers: Vec<String>, pub otlp_headers: Vec<String>,
@ -106,9 +96,8 @@ pub struct DoctorArgs {
pub log_level: LogLevel, pub log_level: LogLevel,
} }
/// Flags shared by the two one-shot subcommands, `merge` and `query`.
#[derive(Debug, clap::Args)] #[derive(Debug, clap::Args)]
pub struct VaultQueryArgs { pub struct MergeArgs {
/// Path to the vault root. /// Path to the vault root.
#[arg(long, env = "ANWESEN_VAULT")] #[arg(long, env = "ANWESEN_VAULT")]
pub vault: PathBuf, pub vault: PathBuf,
@ -116,12 +105,12 @@ pub struct VaultQueryArgs {
/// Query in the `/query` query-string grammar, for example /// Query in the `/query` query-string grammar, for example
/// `tags=anwesen&__anw-kind=skill&__anw-order=order`. The `__anw-kind` /// `tags=anwesen&__anw-kind=skill&__anw-order=order`. The `__anw-kind`
/// homogeneity guard and `__anw-order` fragment ordering ride inside this /// homogeneity guard and `__anw-order` fragment ordering ride inside this
/// string -- there are no separate flags. `__anw-kind` and `__anw-order` /// string -- there are no separate flags. Empty merges every note under
/// apply to `merge` only. Empty matches every note under the vault root. /// the vault root.
#[arg(long, env = "ANWESEN_QUERY", default_value = "")] #[arg(long, env = "ANWESEN_QUERY", default_value = "")]
pub query: String, pub query: String,
/// Log verbosity. Logs go to stderr; the document goes to stdout. /// Log verbosity. Logs go to stderr; the merged document goes to stdout.
#[arg(long, env = "ANWESEN_LOG_LEVEL", default_value = "info")] #[arg(long, env = "ANWESEN_LOG_LEVEL", default_value = "info")]
pub log_level: LogLevel, pub log_level: LogLevel,
} }
@ -183,34 +172,6 @@ mod tests {
} }
} }
/// The removed telemetry flags still parse, hidden from `--help`. Clap
/// rejecting them as unknown would say nothing about the `OTEL_`
/// replacement; the migration error in `telemetry::TelemetryConfig`
/// needs the values to reach it (ANW-42).
#[test]
fn serve_still_parses_the_removed_telemetry_flags() {
let cli = parse(&[
"serve",
"--vault",
"/tmp/v",
"--uptrace-dsn",
"https://tok@api.uptrace.dev",
"--otlp-endpoint",
"https://collector.example.com",
"--otlp-header",
"authorization=Bearer xyz",
])
.expect("parse");
match cli.command {
Command::Serve(a) => {
assert!(a.uptrace_dsn.is_some());
assert!(a.otlp_endpoint.is_some());
assert_eq!(a.otlp_headers, vec!["authorization=Bearer xyz".to_string()]);
}
_ => panic!("expected serve"),
}
}
#[test] #[test]
fn serve_rejects_malformed_bind_at_parse_time() { fn serve_rejects_malformed_bind_at_parse_time() {
let err = parse(&["serve", "--vault", "/tmp/v", "--bind", "not-an-addr"]).unwrap_err(); let err = parse(&["serve", "--vault", "/tmp/v", "--bind", "not-an-addr"]).unwrap_err();
@ -281,47 +242,6 @@ mod tests {
assert_eq!(err.kind(), clap::error::ErrorKind::UnknownArgument); assert_eq!(err.kind(), clap::error::ErrorKind::UnknownArgument);
} }
#[test]
fn query_takes_the_same_flags_as_merge() {
let cli = parse(&[
"query",
"--vault",
"/tmp/v",
"--query",
"tags=anwesen&__anw-limit=5",
"--log-level",
"warn",
])
.expect("parse");
match cli.command {
Command::Query(a) => {
assert_eq!(a.vault, PathBuf::from("/tmp/v"));
assert_eq!(a.query, "tags=anwesen&__anw-limit=5");
assert!(matches!(a.log_level, LogLevel::Warn));
}
_ => panic!("expected query"),
}
}
#[test]
fn query_requires_vault_and_defaults_the_query() {
assert!(parse(&["query"]).is_err());
match parse(&["query", "--vault", "/tmp/v"])
.expect("parse")
.command
{
Command::Query(a) => assert_eq!(a.query, ""),
_ => panic!("expected query"),
}
}
#[test]
fn query_rejects_bind() {
// --bind is serve-only; query does not listen.
let err = parse(&["query", "--vault", "/tmp/v", "--bind", "0.0.0.0:9000"]).unwrap_err();
assert_eq!(err.kind(), clap::error::ErrorKind::UnknownArgument);
}
#[test] #[test]
fn version_takes_no_flags() { fn version_takes_no_flags() {
assert!(matches!( assert!(matches!(

2
src/doctor.rs
View file

@ -46,7 +46,7 @@ pub struct DriftShape {
/// `"bool"`, `"number"`, `"string"`, `"date"`, `"list"`, `"mapping"`. /// `"bool"`, `"number"`, `"string"`, `"date"`, `"list"`, `"mapping"`.
pub shape: &'static str, pub shape: &'static str,
pub count: usize, pub count: usize,
/// Up to `DRIFT_SAMPLES_PER_SHAPE` vault-relative paths exhibiting /// Up to [`DRIFT_SAMPLES_PER_SHAPE`] vault-relative paths exhibiting
/// this shape, in the order they were scanned. /// this shape, in the order they were scanned.
pub samples: Vec<String>, pub samples: Vec<String>,
} }

33
src/http.rs
View file

@ -22,7 +22,7 @@ use axum::http::{HeaderMap, HeaderValue, StatusCode};
use axum::middleware::{Next, from_fn, from_fn_with_state}; use axum::middleware::{Next, from_fn, from_fn_with_state};
use axum::response::{IntoResponse, Response}; use axum::response::{IntoResponse, Response};
use axum::routing::get; use axum::routing::get;
use chrono::{DateTime, Utc}; use chrono::{DateTime, SecondsFormat, Utc};
use http_body::Body as _; use http_body::Body as _;
use hydra::Process; use hydra::Process;
use serde::Serialize; use serde::Serialize;
@ -32,11 +32,17 @@ use std::path::PathBuf;
use crate::app::RestartCounters; use crate::app::RestartCounters;
use crate::health::HealthState; use crate::health::HealthState;
use crate::query::rfc3339_z;
use crate::store::NoteStore; use crate::store::NoteStore;
use crate::telemetry::{self, Telemetry, TraceHeaders}; use crate::telemetry::{self, Telemetry, TraceHeaders};
use crate::vault::{Note, frontmatter_to_json}; use crate::vault::{Note, frontmatter_to_json};
/// Canonical RFC 3339 form with a `Z` suffix -- the shape the User Manual
/// example uses for `last_modified`. Centralized here so every HTTP
/// `last_modified` field stays in the same dialect.
fn rfc3339_z(dt: DateTime<Utc>) -> String {
dt.to_rfc3339_opts(SecondsFormat::Secs, true)
}
/// Shared state injected into every handler. /// Shared state injected into every handler.
#[derive(Clone)] #[derive(Clone)]
pub struct HttpState { pub struct HttpState {
@ -813,23 +819,18 @@ mod tests {
} }
/// With telemetry installed, a normal request is answered byte-for-byte /// With telemetry installed, a normal request is answered byte-for-byte
/// as without it. The test process sets no `OTEL_EXPORTER_OTLP_*` /// as without it. The exporter points at an unreachable local port, so
/// variables, so the exporter aims at the SDK default and fails in the /// export fails instantly in the background and never touches the
/// background without ever touching the response path. /// response path.
#[tokio::test] #[tokio::test]
async fn telemetry_layer_does_not_alter_responses() { async fn telemetry_layer_does_not_alter_responses() {
use crate::telemetry::{self, OtelEnv, RawTelemetryArgs, TelemetryConfig}; use crate::telemetry::{self, RawTelemetryArgs, TelemetryConfig};
let cfg = TelemetryConfig::resolve( let cfg = TelemetryConfig::resolve(RawTelemetryArgs {
&RawTelemetryArgs { otlp_endpoint: Some("http://127.0.0.1:9".into()),
slow_request_ms: 500, slow_request_ms: 500,
..Default::default() ..Default::default()
}, })
OtelEnv {
endpoint: Some("http://127.0.0.1:9".into()),
..OtelEnv::default()
},
)
.unwrap() .unwrap()
.expect("telemetry on"); .expect("telemetry on");
let tel = Arc::new(telemetry::init(cfg).expect("telemetry init")); let tel = Arc::new(telemetry::init(cfg).expect("telemetry init"));

2
src/lib.rs
View file

@ -10,7 +10,7 @@ pub mod app;
pub mod doctor; pub mod doctor;
pub mod health; pub mod health;
pub mod http; pub mod http;
pub mod oneshot; pub mod merge;
pub mod query; pub mod query;
pub mod store; pub mod store;
pub mod telemetry; pub mod telemetry;

56
src/main.rs
View file

@ -1,17 +1,16 @@
//! Anwesen: read-only HTTP daemon over a markdown vault. //! Anwesen: read-only HTTP daemon over a markdown vault.
//! //!
//! This module wires the CLI to the `serve`, `doctor`, `merge`, `query`, and //! This module wires the CLI to the `serve`, `doctor`, `merge`, and
//! `version` subcommands. //! `version` subcommands.
mod cli; mod cli;
use std::sync::Arc; use std::sync::Arc;
use std::sync::atomic::Ordering;
use anwesen::app::Anwesen; use anwesen::app::Anwesen;
use anwesen::doctor; use anwesen::doctor;
use anwesen::oneshot; use anwesen::merge;
use anwesen::telemetry::{self, OtelEnv, RawTelemetryArgs, TelemetryConfig}; use anwesen::telemetry::{self, RawTelemetryArgs, TelemetryConfig};
use anyhow::Result; use anyhow::Result;
use clap::Parser; use clap::Parser;
use hydra::Application; use hydra::Application;
@ -26,18 +25,14 @@ fn main() -> Result<()> {
match cli.command { match cli.command {
Command::Serve(args) => { Command::Serve(args) => {
init_logging(args.log_level); init_logging(args.log_level);
// Resolve telemetry config before the supervisor starts; no // Resolve telemetry config before the supervisor starts; an
// OTEL_EXPORTER_OTLP_* endpoint leaves it `None` (export off, // unset endpoint leaves it `None` (export off, behaves as today).
// no middleware). A removed flag is a startup error (ANW-42). let telemetry = match TelemetryConfig::resolve(RawTelemetryArgs {
let telemetry = match TelemetryConfig::resolve( uptrace_dsn: args.uptrace_dsn,
&RawTelemetryArgs { otlp_endpoint: args.otlp_endpoint,
uptrace_dsn: args.uptrace_dsn, otlp_headers: args.otlp_headers,
otlp_endpoint: args.otlp_endpoint, slow_request_ms: args.otlp_slow_request_ms,
otlp_headers: args.otlp_headers, })? {
slow_request_ms: args.otlp_slow_request_ms,
},
OtelEnv::from_env(),
)? {
Some(cfg) => Some(Arc::new(telemetry::init(cfg)?)), Some(cfg) => Some(Arc::new(telemetry::init(cfg)?)),
None => None, None => None,
}; };
@ -47,22 +42,12 @@ fn main() -> Result<()> {
telemetry = telemetry.is_some(), telemetry = telemetry.is_some(),
"anwesen serve: starting supervisor tree" "anwesen serve: starting supervisor tree"
); );
let app = Anwesen::new(args.vault, args.bind, telemetry.clone());
let started = app.started.clone();
// Blocks until the supervisor exits (SIGTERM / SIGINT / crash). // Blocks until the supervisor exits (SIGTERM / SIGINT / crash).
app.run(); Anwesen::new(args.vault, args.bind, telemetry.clone()).run();
// Flush and shut down exporters after the server loop returns. // Flush and shut down exporters after the server loop returns.
if let Some(telemetry) = telemetry { if let Some(telemetry) = telemetry {
telemetry.shutdown(); telemetry.shutdown();
} }
// `run` returns normally whether the tree came up or never
// started, so a failed start would otherwise look like a clean
// exit and systemd's `Restart=on-failure` would not retry
// (ANW-45). Exit nonzero when the tree never came up.
if !started.load(Ordering::Acquire) {
tracing::error!("anwesen serve: supervisor tree failed to start");
std::process::exit(1);
}
} }
Command::Doctor(args) => { Command::Doctor(args) => {
init_logging(args.log_level); init_logging(args.log_level);
@ -74,26 +59,13 @@ fn main() -> Result<()> {
} }
Command::Merge(args) => { Command::Merge(args) => {
init_logging(args.log_level); init_logging(args.log_level);
match oneshot::merge(&args.vault, &args.query) { match merge::run(&args.vault, &args.query) {
// `print!`, not `println!`: the merged document is byte-stable // `print!`, not `println!`: the merged document is byte-stable
// and byte-identical to the HTTP merge body, which carries no // and byte-identical to the HTTP merge body, which carries no
// trailing newline. An empty match set prints nothing, exit 0. // trailing newline. An empty match set prints nothing, exit 0.
Ok(doc) => print!("{doc}"), Ok(doc) => print!("{doc}"),
Err(e) => { Err(e) => {
eprint!("{}", e.render("merge")); eprint!("{}", e.render());
std::process::exit(1);
}
}
}
Command::Query(args) => {
init_logging(args.log_level);
match oneshot::query_json(&args.vault, &args.query) {
// `println!` here, unlike `merge`: the JSON body carries no
// trailing newline either, but stdout is one document per
// line for the shells and `jq` pipelines this exists for.
Ok(doc) => println!("{doc}"),
Err(e) => {
eprint!("{}", e.render("query"));
std::process::exit(1); std::process::exit(1);
} }
} }

205
src/merge.rs Normal file
View file

@ -0,0 +1,205 @@
//! One-shot local markdown-merge for the `anwesen merge` subcommand ([ANW-27]).
//!
//! Walks a vault directory once (the same [`vault::scan`] as `doctor`),
//! evaluates the `--query` string, and assembles the merged markdown document
//! with [`query::execute_merge`] -- the very engine the HTTP `/query` merge
//! mode ([ANW-26]) uses. CLI and HTTP output are therefore byte-identical for
//! the same vault and query. No HTTP, no watcher, no persistent index.
use std::fmt::Write as _;
use std::path::Path;
use crate::query::{self, MergeError, QueryError};
use crate::store::NoteStore;
use crate::vault::{self, ScanIssue};
/// Why a one-shot merge could not produce a document.
#[derive(Debug)]
pub enum MergeCliError {
/// The `--query` string did not parse. Same grammar, same message as the
/// HTTP `/query` `400`.
Query(QueryError),
/// One or more files could not be read, or their frontmatter did not
/// parse. Same hard-failure posture as `doctor`; soft warnings (e.g. a
/// non-mapping frontmatter root) are ignored, matching what `serve`
/// ingests.
Scan(Vec<ScanIssue>),
/// The `__anw-kind` homogeneity guard rejected the matched set. The
/// `String` is the same naming message the HTTP path returns as `400`,
/// surfaced here on stderr instead.
Kind(String),
}
impl MergeCliError {
/// Render the error for stderr. One concern per line, deterministic so the
/// stderr surface is golden-testable.
#[must_use]
pub fn render(&self) -> String {
match self {
Self::Query(e) => format!("{e}\n"),
Self::Scan(issues) => {
let mut s = String::from("merge: cannot read vault\n");
for issue in issues {
let _ = writeln!(s, " {}: {}", issue.path.display(), issue.kind);
}
s
}
Self::Kind(msg) => msg.clone(),
}
}
}
/// Walk `vault_root`, evaluate `raw_query`, and return the merged document.
///
/// On success the returned `String` is byte-identical to the HTTP merge body
/// for the same vault and query. An empty match set yields an empty string.
///
/// # Errors
/// - [`MergeCliError::Query`] when `raw_query` is malformed;
/// - [`MergeCliError::Scan`] when the directory is unreadable or any file's
/// frontmatter fails to parse;
/// - [`MergeCliError::Kind`] when the `__anw-kind` homogeneity guard fails.
pub fn run(vault_root: &Path, raw_query: &str) -> Result<String, MergeCliError> {
let parsed = query::parse(raw_query).map_err(MergeCliError::Query)?;
let scan = vault::scan(vault_root);
if !scan.issues.is_empty() {
return Err(MergeCliError::Scan(scan.issues));
}
// The merge engine reads from a NoteStore exactly as the HTTP path does;
// a one-shot `replace` is the whole "index" this subcommand needs.
let store = NoteStore::new();
store.replace(scan.notes);
query::execute_merge(&store, &parsed)
.map_err(|MergeError::KindGuard(msg)| MergeCliError::Kind(msg))
}
#[cfg(test)]
mod tests {
use super::*;
use std::fs;
use std::path::Path;
use tempfile::TempDir;
fn write(root: &Path, rel: &str, body: &str) {
let p = root.join(rel);
if let Some(parent) = p.parent() {
fs::create_dir_all(parent).unwrap();
}
fs::write(p, body).unwrap();
}
#[test]
fn merges_bodies_with_source_markers() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nnum: 1\n---\nalpha\n");
write(tmp.path(), "b.md", "---\nnum: 2\n---\nbeta\n");
let out = run(tmp.path(), "").unwrap();
// Body is frontmatter-stripped; fragments join with a blank line and
// there is no trailing newline -- byte-identical to the HTTP body.
assert_eq!(
out,
"<!-- source: a.md -->\nalpha\n\n\n<!-- source: b.md -->\nbeta\n"
);
}
#[test]
fn order_desc_then_path_tiebreak() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nnum: 1\n---\nlow\n");
write(tmp.path(), "b.md", "---\nnum: 3\n---\nhigh\n");
write(tmp.path(), "c.md", "---\nnum: 2\n---\nmid\n");
let out = run(tmp.path(), "__anw-order=num:desc").unwrap();
let bodies: Vec<&str> = out
.lines()
.filter(|l| !l.starts_with("<!--") && !l.is_empty())
.collect();
assert_eq!(bodies, vec!["high", "mid", "low"]);
}
#[test]
fn empty_match_set_is_empty_string() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\ntags: [x]\n---\nbody\n");
// Predicate matches nothing.
assert_eq!(run(tmp.path(), "tags=nope").unwrap(), "");
}
#[test]
fn empty_vault_is_empty_string() {
let tmp = TempDir::new().unwrap();
assert_eq!(run(tmp.path(), "").unwrap(), "");
}
#[test]
fn kind_guard_passes_when_uniform() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nkind: skill\n---\nA\n");
write(tmp.path(), "b.md", "---\nkind: skill\n---\nB\n");
assert!(run(tmp.path(), "__anw-kind=kind").is_ok());
}
#[test]
fn kind_guard_rejects_distinct_values_naming_offenders() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nkind: skill\n---\nA\n");
write(tmp.path(), "b.md", "---\nkind: note\n---\nB\n");
let err = run(tmp.path(), "__anw-kind=kind").unwrap_err();
let msg = err.render();
assert!(msg.contains("distinct values found"));
assert!(msg.contains("a.md"));
assert!(msg.contains("b.md"));
}
#[test]
fn kind_guard_rejects_missing_key() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nkind: skill\n---\nA\n");
write(tmp.path(), "b.md", "---\n---\nB\n");
let err = run(tmp.path(), "__anw-kind=kind").unwrap_err();
assert!(matches!(err, MergeCliError::Kind(_)));
assert!(err.render().contains("notes missing the key"));
}
#[test]
fn malformed_query_is_reported() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\n---\nbody\n");
let err = run(tmp.path(), "x__bogus=1").unwrap_err();
assert!(matches!(err, MergeCliError::Query(_)));
// Non-empty stderr surface.
assert!(!err.render().is_empty());
}
#[test]
fn unparseable_frontmatter_is_a_scan_error() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "bad.md", "---\n:: :: ::\n---\n");
let err = run(tmp.path(), "").unwrap_err();
assert!(matches!(err, MergeCliError::Scan(_)));
assert!(err.render().contains("cannot read vault"));
assert!(err.render().contains("bad.md"));
}
#[test]
fn output_matches_http_engine_for_same_vault_and_query() {
// The shared-engine check: the CLI path and a directly-built store
// feeding execute_merge must produce the identical document.
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nnum: 2\n---\nA body\n");
write(tmp.path(), "b.md", "---\nnum: 1\n---\nB body\n");
let raw = "__anw-order=num:asc";
let cli = run(tmp.path(), raw).unwrap();
let scan = vault::scan(tmp.path());
let store = NoteStore::new();
store.replace(scan.notes);
let parsed = query::parse(raw).unwrap();
let direct = query::execute_merge(&store, &parsed).unwrap();
assert_eq!(cli, direct);
}
}

354
src/oneshot.rs
View file

@ -1,354 +0,0 @@
//! One-shot local query evaluation for the `anwesen merge` ([ANW-27]) and
//! `anwesen query` ([ANW-43]) subcommands.
//!
//! Both walk a vault directory once (the same [`vault::scan`] as `doctor`),
//! evaluate the `--query` string, and hand the resulting store to the very
//! engine the HTTP `/query` endpoint uses: [`query::execute_merge`] for the
//! merged markdown document ([ANW-26]), [`query::execute`] for the JSON
//! projection. CLI and HTTP output are therefore byte-identical for the same
//! vault and query. No HTTP, no watcher, no persistent index.
use std::fmt::Write as _;
use std::path::Path;
use std::sync::Arc;
use crate::query::{self, MergeError, ParsedQuery, QueryError};
use crate::store::NoteStore;
use crate::vault::{self, ScanIssue};
/// Why a one-shot evaluation could not produce a document.
#[derive(Debug)]
pub enum OneshotError {
/// The `--query` string did not parse. Same grammar, same message as the
/// HTTP `/query` `400`.
Query(QueryError),
/// One or more files could not be read, or their frontmatter did not
/// parse. Same hard-failure posture as `doctor`; soft warnings (e.g. a
/// non-mapping frontmatter root) are ignored, matching what `serve`
/// ingests.
Scan(Vec<ScanIssue>),
/// The `__anw-kind` homogeneity guard rejected the matched set. The
/// `String` is the same naming message the HTTP path returns as `400`,
/// surfaced here on stderr instead. Merge mode only -- the JSON
/// projection does not run the guard, exactly as the endpoint does not.
Kind(String),
}
impl OneshotError {
/// Render the error for stderr. One concern per line, deterministic so the
/// stderr surface is golden-testable. `subcommand` names the caller so the
/// scan header reads `merge:` or `query:` as the operator invoked it.
#[must_use]
pub fn render(&self, subcommand: &str) -> String {
match self {
Self::Query(e) => format!("{e}\n"),
Self::Scan(issues) => {
let mut s = format!("{subcommand}: cannot read vault\n");
for issue in issues {
let _ = writeln!(s, " {}: {}", issue.path.display(), issue.kind);
}
s
}
Self::Kind(msg) => msg.clone(),
}
}
}
/// Parse `raw_query` and load `vault_root` into a one-shot store.
///
/// The single path both subcommands take, so the two cannot drift in how they
/// parse a query or how strictly they read a vault.
fn load(vault_root: &Path, raw_query: &str) -> Result<(ParsedQuery, Arc<NoteStore>), OneshotError> {
let parsed = query::parse(raw_query).map_err(OneshotError::Query)?;
let scan = vault::scan(vault_root);
if !scan.issues.is_empty() {
return Err(OneshotError::Scan(scan.issues));
}
// The engine reads from a NoteStore exactly as the HTTP path does; a
// one-shot `replace` is the whole "index" these subcommands need.
let store = NoteStore::new();
store.replace(scan.notes);
Ok((parsed, store))
}
/// Walk `vault_root`, evaluate `raw_query`, and return the merged document.
///
/// On success the returned `String` is byte-identical to the HTTP merge body
/// for the same vault and query. An empty match set yields an empty string.
///
/// # Errors
/// - [`OneshotError::Query`] when `raw_query` is malformed;
/// - [`OneshotError::Scan`] when the directory is unreadable or any file's
/// frontmatter fails to parse;
/// - [`OneshotError::Kind`] when the `__anw-kind` homogeneity guard fails.
pub fn merge(vault_root: &Path, raw_query: &str) -> Result<String, OneshotError> {
let (parsed, store) = load(vault_root, raw_query)?;
query::execute_merge(&store, &parsed)
.map_err(|MergeError::KindGuard(msg)| OneshotError::Kind(msg))
}
/// Walk `vault_root`, evaluate `raw_query`, and return the JSON document
/// `GET /query` returns for the same vault and query -- same projection, same
/// `results` / `total` / `truncated` shape, same compact serialization, so a
/// script moving between daemon and CLI needs no second parser ([ANW-43]).
///
/// Result ordering and the `__anw-limit` semantics are the endpoint's,
/// because this is the endpoint's [`query::execute`]: paths ascending, the
/// cap applied after `total` is counted.
///
/// # Errors
/// - [`OneshotError::Query`] when `raw_query` is malformed;
/// - [`OneshotError::Scan`] when the directory is unreadable or any file's
/// frontmatter fails to parse.
///
/// # Panics
/// If the response fails to serialize -- unreachable, as the HTTP handler's
/// identical `to_vec` call also treats it: every field is a string, a number,
/// or frontmatter already converted to `serde_json::Value`.
pub fn query_json(vault_root: &Path, raw_query: &str) -> Result<String, OneshotError> {
let (parsed, store) = load(vault_root, raw_query)?;
let response = query::execute(&store, &parsed, query::rfc3339_z);
Ok(serde_json::to_string(&response).expect("query response serializes"))
}
#[cfg(test)]
mod tests {
use super::*;
use std::fs;
use std::path::Path;
use tempfile::TempDir;
fn write(root: &Path, rel: &str, body: &str) {
let p = root.join(rel);
if let Some(parent) = p.parent() {
fs::create_dir_all(parent).unwrap();
}
fs::write(p, body).unwrap();
}
#[test]
fn merges_bodies_with_source_markers() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nnum: 1\n---\nalpha\n");
write(tmp.path(), "b.md", "---\nnum: 2\n---\nbeta\n");
let out = merge(tmp.path(), "").unwrap();
// Body is frontmatter-stripped; fragments join with a blank line and
// there is no trailing newline -- byte-identical to the HTTP body.
assert_eq!(
out,
"<!-- source: a.md -->\nalpha\n\n\n<!-- source: b.md -->\nbeta\n"
);
}
#[test]
fn order_desc_then_path_tiebreak() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nnum: 1\n---\nlow\n");
write(tmp.path(), "b.md", "---\nnum: 3\n---\nhigh\n");
write(tmp.path(), "c.md", "---\nnum: 2\n---\nmid\n");
let out = merge(tmp.path(), "__anw-order=num:desc").unwrap();
let bodies: Vec<&str> = out
.lines()
.filter(|l| !l.starts_with("<!--") && !l.is_empty())
.collect();
assert_eq!(bodies, vec!["high", "mid", "low"]);
}
#[test]
fn empty_match_set_is_empty_string() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\ntags: [x]\n---\nbody\n");
// Predicate matches nothing.
assert_eq!(merge(tmp.path(), "tags=nope").unwrap(), "");
}
#[test]
fn empty_vault_is_empty_string() {
let tmp = TempDir::new().unwrap();
assert_eq!(merge(tmp.path(), "").unwrap(), "");
}
#[test]
fn kind_guard_passes_when_uniform() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nkind: skill\n---\nA\n");
write(tmp.path(), "b.md", "---\nkind: skill\n---\nB\n");
assert!(merge(tmp.path(), "__anw-kind=kind").is_ok());
}
#[test]
fn kind_guard_rejects_distinct_values_naming_offenders() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nkind: skill\n---\nA\n");
write(tmp.path(), "b.md", "---\nkind: note\n---\nB\n");
let err = merge(tmp.path(), "__anw-kind=kind").unwrap_err();
let msg = err.render("merge");
assert!(msg.contains("distinct values found"));
assert!(msg.contains("a.md"));
assert!(msg.contains("b.md"));
}
#[test]
fn kind_guard_rejects_missing_key() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nkind: skill\n---\nA\n");
write(tmp.path(), "b.md", "---\n---\nB\n");
let err = merge(tmp.path(), "__anw-kind=kind").unwrap_err();
assert!(matches!(err, OneshotError::Kind(_)));
assert!(err.render("merge").contains("notes missing the key"));
}
#[test]
fn malformed_query_is_reported() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\n---\nbody\n");
let err = merge(tmp.path(), "x__bogus=1").unwrap_err();
assert!(matches!(err, OneshotError::Query(_)));
// Non-empty stderr surface.
assert!(!err.render("merge").is_empty());
}
#[test]
fn unparseable_frontmatter_is_a_scan_error() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "bad.md", "---\n:: :: ::\n---\n");
let err = merge(tmp.path(), "").unwrap_err();
assert!(matches!(err, OneshotError::Scan(_)));
assert!(err.render("merge").contains("cannot read vault"));
assert!(err.render("merge").contains("bad.md"));
}
#[test]
fn output_matches_http_engine_for_same_vault_and_query() {
// The shared-engine check: the CLI path and a directly-built store
// feeding execute_merge must produce the identical document.
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nnum: 2\n---\nA body\n");
write(tmp.path(), "b.md", "---\nnum: 1\n---\nB body\n");
let raw = "__anw-order=num:asc";
let cli = merge(tmp.path(), raw).unwrap();
let scan = vault::scan(tmp.path());
let store = NoteStore::new();
store.replace(scan.notes);
let parsed = query::parse(raw).unwrap();
let direct = query::execute_merge(&store, &parsed).unwrap();
assert_eq!(cli, direct);
}
// --- `query` (ANW-43) --------------------------------------------------
fn json(doc: &str) -> serde_json::Value {
serde_json::from_str(doc).expect("query output parses as JSON")
}
#[test]
fn query_projects_path_frontmatter_and_file_facts() {
let tmp = TempDir::new().unwrap();
write(
tmp.path(),
"Projects/alpha.md",
"---\ntags: [project]\nversion: 2\n---\nbody\n",
);
let doc = json(&query_json(tmp.path(), "").unwrap());
assert_eq!(doc["total"], 1);
assert_eq!(doc["truncated"], false);
let row = &doc["results"][0];
assert_eq!(row["path"], "Projects/alpha.md");
assert_eq!(row["frontmatter"]["version"], 2);
assert_eq!(row["frontmatter"]["tags"][0], "project");
assert_eq!(row["size"], 40);
// The endpoint's timestamp dialect, seconds precision with a Z suffix.
let lm = row["last_modified"].as_str().unwrap();
assert!(lm.ends_with('Z'), "{lm}");
assert!(row["etag"].as_str().unwrap().starts_with('"'));
// The body is elided here as it is on the endpoint.
assert!(row.get("body").is_none());
}
#[test]
fn query_predicates_filter_and_limit_truncates_after_total() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nstatus: draft\n---\nA\n");
write(tmp.path(), "b.md", "---\nstatus: draft\n---\nB\n");
write(tmp.path(), "c.md", "---\nstatus: done\n---\nC\n");
let filtered = json(&query_json(tmp.path(), "status=draft").unwrap());
assert_eq!(filtered["total"], 2);
assert_eq!(filtered["results"].as_array().unwrap().len(), 2);
let capped = json(&query_json(tmp.path(), "status=draft&__anw-limit=1").unwrap());
// `total` counts the full match set; the cap keeps the first row.
assert_eq!(capped["total"], 2);
assert_eq!(capped["truncated"], true);
assert_eq!(capped["results"].as_array().unwrap().len(), 1);
assert_eq!(capped["results"][0]["path"], "a.md");
}
#[test]
fn query_empty_match_set_is_an_empty_result_list() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\ntags: [x]\n---\nbody\n");
let doc = json(&query_json(tmp.path(), "tags=nope").unwrap());
assert_eq!(doc["total"], 0);
assert_eq!(doc["truncated"], false);
assert!(doc["results"].as_array().unwrap().is_empty());
}
#[test]
fn query_malformed_query_is_reported_naming_the_subcommand() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\n---\nbody\n");
let err = query_json(tmp.path(), "x__bogus=1").unwrap_err();
assert!(matches!(err, OneshotError::Query(_)));
assert!(!err.render("query").is_empty());
}
#[test]
fn query_unparseable_frontmatter_is_a_scan_error() {
let tmp = TempDir::new().unwrap();
write(tmp.path(), "bad.md", "---\n:: :: ::\n---\n");
let err = query_json(tmp.path(), "").unwrap_err();
assert!(err.render("query").starts_with("query: cannot read vault"));
assert!(err.render("query").contains("bad.md"));
}
#[test]
fn query_kind_guard_does_not_apply() {
// `__anw-kind` guards merge output only. Parsing it must not make the
// JSON path fail on a mixed set, matching the endpoint.
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nkind: skill\n---\nA\n");
write(tmp.path(), "b.md", "---\nkind: note\n---\nB\n");
assert!(merge(tmp.path(), "__anw-kind=kind").is_err());
let doc = json(&query_json(tmp.path(), "__anw-kind=kind").unwrap());
assert_eq!(doc["total"], 2);
}
#[test]
fn query_output_matches_the_http_projection_byte_for_byte() {
// The shared-engine check for the JSON surface: same store, same
// `execute`, same serialization as the `/query` handler.
let tmp = TempDir::new().unwrap();
write(tmp.path(), "a.md", "---\nnum: 2\n---\nA body\n");
write(tmp.path(), "b.md", "---\nnum: 1\n---\nB body\n");
let raw = "__anw-limit=1";
let cli = query_json(tmp.path(), raw).unwrap();
let scan = vault::scan(tmp.path());
let store = NoteStore::new();
store.replace(scan.notes);
let parsed = query::parse(raw).unwrap();
let direct =
serde_json::to_string(&query::execute(&store, &parsed, crate::query::rfc3339_z))
.unwrap();
assert_eq!(cli, direct);
}
}

14
src/query.rs
View file

@ -11,14 +11,14 @@
//! `__in` / `__all` are comma-separated; unknown operators are `400`. //! `__in` / `__all` are comma-separated; unknown operators are `400`.
//! //!
//! Predicates are evaluated by iterating the in-memory [`NoteStore`] and //! Predicates are evaluated by iterating the in-memory [`NoteStore`] and
//! applying each [`Predicate`] in turn. At the documented scale //! applying each [`Predicate::matches`] in turn. At the documented scale
//! (low-thousands-of-notes vaults) this is sub-millisecond; see //! (low-thousands-of-notes vaults) this is sub-millisecond; see
//! [[ADR-009 Reverse ADR-002 In-Memory Evaluation No Tantivy]] for the //! [[ADR-009 Reverse ADR-002 In-Memory Evaluation No Tantivy]] for the
//! call to keep evaluation in-memory rather than carrying a Tantivy index. //! call to keep evaluation in-memory rather than carrying a Tantivy index.
use std::collections::BTreeMap; use std::collections::BTreeMap;
use chrono::{DateTime, NaiveDate, SecondsFormat, Utc}; use chrono::{DateTime, NaiveDate};
use regex::Regex; use regex::Regex;
use serde::Serialize; use serde::Serialize;
use serde_json::Value as JsonValue; use serde_json::Value as JsonValue;
@ -280,16 +280,6 @@ fn hex_value(b: u8) -> Option<u8> {
} }
} }
/// Canonical RFC 3339 form with a `Z` suffix -- the shape the User Manual
/// example uses for `last_modified`. It lives next to [`ResultEntry`] because
/// it is the projection's dialect: both the HTTP handlers and the offline
/// `anwesen query` subcommand ([ANW-43]) format timestamps with it, so the
/// two surfaces cannot drift.
#[must_use]
pub fn rfc3339_z(dt: DateTime<Utc>) -> String {
dt.to_rfc3339_opts(SecondsFormat::Secs, true)
}
/// One result row in the `/query` response. Per User Manual the body is /// One result row in the `/query` response. Per User Manual the body is
/// elided -- consumers fetch bodies via `/notes/<path>` if needed. /// elided -- consumers fetch bodies via `/notes/<path>` if needed.
#[derive(Debug, Serialize)] #[derive(Debug, Serialize)]

623
src/telemetry.rs
View file

@ -12,288 +12,140 @@
//! the propagated context. Every other request stays metrics-only, so the //! the propagated context. Every other request stays metrics-only, so the
//! ~3.6k requests/min steady state does not drown the trace backend. //! ~3.6k requests/min steady state does not drown the trace backend.
//! //!
//! Transport configuration comes from the standard `OTEL_EXPORTER_OTLP_*` //! When no OTLP endpoint (or uptrace DSN) is configured, [`init`] returns
//! environment variables only, read by the `OpenTelemetry` SDK itself
//! ([ANW-42](https://crvrs.youtrack.cloud/issue/ANW-42)). anwesen parses no
//! addresses and appends no per-signal paths. It checks two things at
//! startup, both cases the SDK would otherwise export nowhere in silence:
//!
//! - `OTEL_EXPORTER_OTLP_ENDPOINT` carries no query or fragment, because the
//! SDK's own concatenation mangles those (measured against a sink: a
//! `?tail` base sends `POST /?tail/v1/metrics`, a `#frag` base sends
//! `POST /`).
//! - No protocol variable asks for anything but `http/protobuf`, the only
//! transport this binary is built with.
//!
//! When no endpoint variable is set, [`TelemetryConfig::resolve`] returns
//! `None`, the request middleware is not installed, and the server behaves //! `None`, the request middleware is not installed, and the server behaves
//! exactly as it did before this module existed. External installs run //! exactly as it did before this module existed. External installs run
//! unchanged. //! unchanged.
//!
//! Config mirrors the gestell uptrace surface (see the gestell PDR-GES-136
//! `[otel]` section): a `uptrace_dsn` shorthand, or a generic endpoint plus
//! headers.
use std::collections::HashMap;
use std::time::{Duration, SystemTime}; use std::time::{Duration, SystemTime};
use anyhow::{Context as _, bail}; use anyhow::{Context as _, anyhow, bail};
use axum::http::HeaderMap; use axum::http::HeaderMap;
use opentelemetry::KeyValue; use opentelemetry::KeyValue;
use opentelemetry::metrics::{Counter, Histogram, MeterProvider as _}; use opentelemetry::metrics::{Counter, Histogram, MeterProvider as _};
use opentelemetry::propagation::{Extractor, TextMapPropagator}; use opentelemetry::propagation::{Extractor, TextMapPropagator};
use opentelemetry::trace::{Span, SpanKind, Tracer, TracerProvider as _}; use opentelemetry::trace::{Span, SpanKind, Tracer, TracerProvider as _};
use opentelemetry_otlp::{MetricExporter, SpanExporter}; use opentelemetry_otlp::{
MetricExporter, Protocol, SpanExporter, WithExportConfig, WithHttpConfig,
};
use opentelemetry_sdk::Resource; use opentelemetry_sdk::Resource;
use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
use opentelemetry_sdk::propagation::TraceContextPropagator; use opentelemetry_sdk::propagation::TraceContextPropagator;
use opentelemetry_sdk::trace::Sampler; use opentelemetry_sdk::trace::Sampler;
use opentelemetry_sdk::trace::{SdkTracer, SdkTracerProvider}; use opentelemetry_sdk::trace::{SdkTracer, SdkTracerProvider};
use opentelemetry_semantic_conventions::resource::{SERVICE_NAME, SERVICE_VERSION}; use opentelemetry_semantic_conventions::resource::SERVICE_VERSION;
/// The removed telemetry options, still parsed so their presence is an error /// Raw telemetry options as parsed by clap on the `serve` command. Resolved
/// rather than a silent config drop on upgrade /// into an [`Option<TelemetryConfig>`] by [`TelemetryConfig::resolve`].
/// ([ANW-42](https://crvrs.youtrack.cloud/issue/ANW-42)), plus the one
/// surviving option. Resolved into an [`Option<TelemetryConfig>`] by
/// [`TelemetryConfig::resolve`].
#[derive(Debug, Default)] #[derive(Debug, Default)]
pub struct RawTelemetryArgs { pub struct RawTelemetryArgs {
/// Removed `--uptrace-dsn` / `ANWESEN_UPTRACE_DSN`. /// `--uptrace-dsn` / `ANWESEN_UPTRACE_DSN`.
pub uptrace_dsn: Option<String>, pub uptrace_dsn: Option<String>,
/// Removed `--otlp-endpoint` / `ANWESEN_OTLP_ENDPOINT`. /// `--otlp-endpoint` / `ANWESEN_OTLP_ENDPOINT`.
pub otlp_endpoint: Option<String>, pub otlp_endpoint: Option<String>,
/// Removed `--otlp-header` / `ANWESEN_OTLP_HEADERS`. /// `--otlp-header` / `ANWESEN_OTLP_HEADERS`, each `key=value`.
pub otlp_headers: Vec<String>, pub otlp_headers: Vec<String>,
/// `--otlp-slow-request-ms` / `ANWESEN_OTLP_SLOW_REQUEST_MS`. Kept: it /// `--otlp-slow-request-ms` / `ANWESEN_OTLP_SLOW_REQUEST_MS`.
/// decides when anwesen emits a span, not where the export goes, and no
/// standard variable covers it.
pub slow_request_ms: u64, pub slow_request_ms: u64,
} }
/// The OTLP transport variables the SDK reads, captured once at startup so /// A resolved, telemetry-on configuration. Built only when an endpoint or a
/// the telemetry-on decision and the startup checks stay pure functions of /// DSN is present; absence is represented by `Ok(None)` from [`resolve`].
/// them. Values are the raw strings; anwesen does not parse them beyond the
/// checks in [`check_generic_endpoint`] and [`check_protocols`].
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct OtelEnv {
/// `OTEL_EXPORTER_OTLP_ENDPOINT`: base URL, per-signal path appended by
/// the SDK.
pub endpoint: Option<String>,
/// `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT`: full URL, used verbatim.
pub metrics_endpoint: Option<String>,
/// `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`: full URL, used verbatim.
pub traces_endpoint: Option<String>,
/// `OTEL_EXPORTER_OTLP_PROTOCOL`.
pub protocol: Option<String>,
/// `OTEL_EXPORTER_OTLP_METRICS_PROTOCOL`.
pub metrics_protocol: Option<String>,
/// `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL`.
pub traces_protocol: Option<String>,
}
impl OtelEnv {
/// Read the transport variables from the process environment. An empty or
/// whitespace-only value counts as unset: an empty endpoint would
/// otherwise turn telemetry on and export to the SDK's `localhost:4318`
/// default.
#[must_use]
pub fn from_env() -> Self {
let var = |name: &str| {
std::env::var(name)
.ok()
.map(|v| v.trim().to_string())
.filter(|v| !v.is_empty())
};
Self {
endpoint: var("OTEL_EXPORTER_OTLP_ENDPOINT"),
metrics_endpoint: var("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"),
traces_endpoint: var("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"),
protocol: var("OTEL_EXPORTER_OTLP_PROTOCOL"),
metrics_protocol: var("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL"),
traces_protocol: var("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL"),
}
}
/// Whether any endpoint variable is set. No endpoint means telemetry off.
fn any_endpoint(&self) -> bool {
self.endpoint.is_some() || self.metrics_endpoint.is_some() || self.traces_endpoint.is_some()
}
/// The endpoint to name in the startup log: the generic base when set,
/// otherwise whichever per-signal URL is.
fn describe(&self) -> &str {
self.endpoint
.as_deref()
.or(self.metrics_endpoint.as_deref())
.or(self.traces_endpoint.as_deref())
.unwrap_or("")
}
}
/// A resolved, telemetry-on configuration. Built only when an endpoint
/// variable is set; absence is represented by `Ok(None)` from [`resolve`].
/// It carries no transport settings: the exporters read those from the
/// environment themselves.
/// ///
/// [`resolve`]: TelemetryConfig::resolve /// [`resolve`]: TelemetryConfig::resolve
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
pub struct TelemetryConfig { pub struct TelemetryConfig {
/// OTLP/HTTP base URL. The exporter appends the per-signal path
/// (`/v1/metrics`, `/v1/traces`).
pub endpoint: String,
/// Export headers (for uptrace, the `uptrace-dsn` entry) as ordered
/// `(name, value)` pairs.
pub headers: Vec<(String, String)>,
/// A request at or over this duration, or answering a 5xx, is recorded /// A request at or over this duration, or answering a 5xx, is recorded
/// as a server span. /// as a server span.
pub slow_request: Duration, pub slow_request: Duration,
/// The transport variables, kept for the startup log line only.
env: OtelEnv,
} }
impl TelemetryConfig { impl TelemetryConfig {
/// Resolve the surviving option plus the OTLP transport variables into an /// Resolve raw clap options into an optional config.
/// optional config.
/// ///
/// - A removed flag or `ANWESEN_` variable is an error naming its `OTEL_` /// - Both `uptrace_dsn` and `otlp_endpoint` set is an error (they are
/// replacement: a deployment exporting today must fail one restart /// two ways to name the same endpoint).
/// rather than go quiet. /// - Neither set means telemetry is off: `Ok(None)`.
/// - No endpoint variable set means telemetry is off: `Ok(None)`. /// - A malformed `key=value` header or an unparseable DSN is an error,
/// - A query or fragment on `OTEL_EXPORTER_OTLP_ENDPOINT` is an error. /// surfaced at startup rather than silently dropping export.
/// - A protocol this binary cannot speak is an error.
/// ///
/// # Errors /// # Errors
/// Returns an error when a removed option is present, when the generic /// Returns an error when the two endpoint sources conflict, a header is
/// endpoint carries a query or a fragment, or when a protocol variable /// not `key=value`, or the uptrace DSN cannot be parsed.
/// asks for anything but `http/protobuf`. pub fn resolve(raw: RawTelemetryArgs) -> anyhow::Result<Option<Self>> {
pub fn resolve(raw: &RawTelemetryArgs, env: OtelEnv) -> anyhow::Result<Option<Self>> { let slow_request = Duration::from_millis(raw.slow_request_ms);
check_removed(raw)?; let mut headers = parse_headers(&raw.otlp_headers)?;
if !env.any_endpoint() {
return Ok(None); match (raw.uptrace_dsn, raw.otlp_endpoint) {
(Some(_), Some(_)) => {
bail!("--uptrace-dsn and --otlp-endpoint are mutually exclusive");
}
(Some(dsn), None) => {
let (endpoint, dsn_header) = parse_uptrace_dsn(&dsn)?;
// The DSN header leads; any explicit --otlp-header follows.
headers.insert(0, dsn_header);
Ok(Some(Self {
endpoint,
headers,
slow_request,
}))
}
(None, Some(endpoint)) => Ok(Some(Self {
endpoint,
headers,
slow_request,
})),
(None, None) => Ok(None),
} }
if let Some(endpoint) = &env.endpoint {
check_generic_endpoint(endpoint)?;
}
check_protocols(&env)?;
Ok(Some(Self {
slow_request: Duration::from_millis(raw.slow_request_ms),
env,
}))
} }
} }
/// Fail on any removed telemetry option, naming the `OTEL_` variable that /// Parse `key=value` header specs. Whitespace around key and value is
/// replaces it. Silence is the expensive failure here: an upgrade that drops /// trimmed; an empty key or a spec with no `=` is an error.
/// the export config would stop telemetry with nothing in the log. fn parse_headers(specs: &[String]) -> anyhow::Result<Vec<(String, String)>> {
fn check_removed(raw: &RawTelemetryArgs) -> anyhow::Result<()> { let mut out = Vec::with_capacity(specs.len());
let removed: [(&str, &str, bool); 3] = [ for spec in specs {
( let (k, v) = spec
"--uptrace-dsn / ANWESEN_UPTRACE_DSN", .split_once('=')
"OTEL_EXPORTER_OTLP_ENDPOINT=https://api.uptrace.dev plus \ .ok_or_else(|| anyhow!("OTLP header {spec:?} is not key=value"))?;
OTEL_EXPORTER_OTLP_HEADERS=uptrace-dsn=<the DSN, verbatim>", let k = k.trim();
raw.uptrace_dsn.is_some(), if k.is_empty() {
), bail!("OTLP header {spec:?} has an empty key");
(
"--otlp-endpoint / ANWESEN_OTLP_ENDPOINT",
"OTEL_EXPORTER_OTLP_ENDPOINT",
raw.otlp_endpoint.is_some(),
),
(
"--otlp-header / ANWESEN_OTLP_HEADERS",
"OTEL_EXPORTER_OTLP_HEADERS",
!raw.otlp_headers.is_empty(),
),
];
for (option, replacement, present) in removed {
if present {
bail!("{option} was removed in anwesen 0.3.0; use {replacement} instead");
} }
out.push((k.to_string(), v.trim().to_string()));
} }
Ok(()) Ok(out)
} }
/// Reject a query or fragment on `OTEL_EXPORTER_OTLP_ENDPOINT`. The SDK /// Parse an uptrace DSN (`https://<token>@host[:port]`) into the OTLP
/// appends the per-signal path to this value textually, so a `?grpc=4317` /// endpoint base URL and the `uptrace-dsn` header uptrace expects. The full
/// tail sends `POST /?grpc=4317/v1/metrics` and a `#frag` tail sends /// DSN is echoed as the header value per uptrace's ingest contract.
/// `POST /` -- both dead exports, neither logged. Measured against a sink on fn parse_uptrace_dsn(dsn: &str) -> anyhow::Result<(String, (String, String))> {
/// opentelemetry-otlp 0.32 ([ANW-42]). A path prefix composes correctly and let dsn = dsn.trim();
/// is left alone. let (scheme, rest) = dsn
/// .split_once("://")
/// [ANW-42]: https://crvrs.youtrack.cloud/issue/ANW-42 .context("uptrace DSN has no scheme (expected https://<token>@host)")?;
fn check_generic_endpoint(endpoint: &str) -> anyhow::Result<()> { // Host is whatever follows the credentials `@`; a DSN with no `@` is
if let Some(bad) = endpoint.find(['?', '#']) { // treated as endpoint-only (lenient, though real uptrace DSNs carry a
let tail = &endpoint[bad..]; // token).
bail!( let host = rest.rsplit_once('@').map_or(rest, |(_, h)| h);
"OTEL_EXPORTER_OTLP_ENDPOINT {endpoint:?} has a trailing {tail:?}; \ let host = host.trim_end_matches('/');
the exporter would append the signal path after it and export nowhere. \ if host.is_empty() {
Pass the base URL alone, and put an uptrace DSN in \ bail!("uptrace DSN has no host");
OTEL_EXPORTER_OTLP_HEADERS=uptrace-dsn=<the DSN, verbatim>"
);
} }
Ok(()) let endpoint = format!("{scheme}://{host}");
} Ok((endpoint, ("uptrace-dsn".to_string(), dsn.to_string())))
/// The one OTLP protocol this binary speaks. `opentelemetry-otlp` is built
/// with the `http-proto` feature alone (Cargo.toml), so neither `grpc` nor
/// `http/json` has a transport behind it.
const SUPPORTED_PROTOCOL: &str = "http/protobuf";
/// Reject a protocol variable this binary cannot honor. The exporter builder
/// picks its transport at compile time, so `OTEL_EXPORTER_OTLP_PROTOCOL=grpc`
/// does not switch anything: the export keeps going out as HTTP protobuf,
/// with nothing in the log (measured, [ANW-42]). An operator who follows
/// uptrace's console to the gRPC port would get exactly the dead-silent
/// export this issue exists to remove, so it fails at startup instead.
///
/// [ANW-42]: https://crvrs.youtrack.cloud/issue/ANW-42
fn check_protocols(env: &OtelEnv) -> anyhow::Result<()> {
let vars = [
("OTEL_EXPORTER_OTLP_PROTOCOL", env.protocol.as_deref()),
(
"OTEL_EXPORTER_OTLP_METRICS_PROTOCOL",
env.metrics_protocol.as_deref(),
),
(
"OTEL_EXPORTER_OTLP_TRACES_PROTOCOL",
env.traces_protocol.as_deref(),
),
];
for (name, value) in vars {
let Some(value) = value else { continue };
if value != SUPPORTED_PROTOCOL {
bail!(
"{name}={value:?} is not supported; this build speaks \
{SUPPORTED_PROTOCOL} only. Unset the variable, and point the \
endpoint at the collector's HTTP port rather than its gRPC one"
);
}
}
Ok(())
}
/// The resource attributes anwesen supplies as defaults, minus every key the
/// environment already sets.
///
/// `service.name=anwesen` and the crate version are defaults, not policy:
/// contract point 6 of [ANW-42] keeps `OTEL_SERVICE_NAME` and
/// `OTEL_RESOURCE_ATTRIBUTES` working. Attaching them unconditionally blocks
/// the override, because an attribute set on the builder wins over the one
/// the SDK's own detectors read from those variables (measured by sie).
///
/// Only key presence matters here; the values stay the SDK's to parse.
///
/// [ANW-42]: https://crvrs.youtrack.cloud/issue/ANW-42
fn default_resource_attrs(
service_name: Option<&str>,
resource_attributes: Option<&str>,
) -> Vec<KeyValue> {
let from_attrs: Vec<&str> = resource_attributes
.unwrap_or_default()
.split(',')
.filter_map(|entry| entry.split_once('='))
.map(|(key, _)| key.trim())
.collect();
let set_by_env = |key: &str| {
(key == SERVICE_NAME && service_name.is_some_and(|v| !v.trim().is_empty()))
|| from_attrs.contains(&key)
};
[
(SERVICE_NAME, "anwesen"),
(SERVICE_VERSION, env!("CARGO_PKG_VERSION")),
]
.into_iter()
.filter(|(key, _)| !set_by_env(key))
.map(|(key, value)| KeyValue::new(key, value))
.collect()
} }
/// Semantic route bucket for the `http.route` label. Coarser than the axum /// Semantic route bucket for the `http.route` label. Coarser than the axum
@ -480,29 +332,28 @@ struct TelemetryInner {
impl TelemetryInner { impl TelemetryInner {
fn new(config: TelemetryConfig) -> anyhow::Result<Self> { fn new(config: TelemetryConfig) -> anyhow::Result<Self> {
let TelemetryConfig { slow_request, env } = config; let TelemetryConfig {
endpoint,
headers,
slow_request,
} = config;
// `.with_endpoint()` is used verbatim by the exporter (the `/v1/*`
// suffix is only auto-appended for the generic OTEL env var), so we
// append the per-signal path ourselves.
let endpoint = endpoint.trim_end_matches('/').to_string();
let header_map: HashMap<String, String> = headers.into_iter().collect();
// Service identity is a default only: a key OTEL_SERVICE_NAME or let resource = Resource::builder()
// OTEL_RESOURCE_ATTRIBUTES already carries is left to the SDK's own .with_service_name("anwesen")
// detectors, because a builder attribute would win over them. .with_attribute(KeyValue::new(SERVICE_VERSION, env!("CARGO_PKG_VERSION")))
let mut resource = Resource::builder(); .build();
for attr in default_resource_attrs(
std::env::var("OTEL_SERVICE_NAME").ok().as_deref(),
std::env::var("OTEL_RESOURCE_ATTRIBUTES").ok().as_deref(),
) {
resource = resource.with_attribute(attr);
}
let resource = resource.build();
// -- metrics: thread-based periodic reader over an OTLP/HTTP exporter. // -- metrics: thread-based periodic reader over an OTLP/HTTP exporter.
// No endpoint, headers, or protocol here: the SDK reads
// OTEL_EXPORTER_OTLP_* itself and composes the per-signal URL
// (ANW-42). `.with_http()` picks the only transport this binary is
// built with, http/protobuf -- the default, and what anwesen sent
// before. Any other OTEL_EXPORTER_OTLP_PROTOCOL value would be
// ignored here, so `check_protocols` rejects it at startup.
let metric_exporter = MetricExporter::builder() let metric_exporter = MetricExporter::builder()
.with_http() .with_http()
.with_protocol(Protocol::HttpBinary)
.with_endpoint(format!("{endpoint}/v1/metrics"))
.with_headers(header_map.clone())
.build() .build()
.context("build OTLP metric exporter")?; .context("build OTLP metric exporter")?;
let reader = PeriodicReader::builder(metric_exporter) let reader = PeriodicReader::builder(metric_exporter)
@ -535,6 +386,9 @@ impl TelemetryInner {
// -- traces: thread-based batch processor over an OTLP/HTTP exporter. // -- traces: thread-based batch processor over an OTLP/HTTP exporter.
let span_exporter = SpanExporter::builder() let span_exporter = SpanExporter::builder()
.with_http() .with_http()
.with_protocol(Protocol::HttpBinary)
.with_endpoint(format!("{endpoint}/v1/traces"))
.with_headers(header_map)
.build() .build()
.context("build OTLP span exporter")?; .context("build OTLP span exporter")?;
let tracer_provider = SdkTracerProvider::builder() let tracer_provider = SdkTracerProvider::builder()
@ -547,7 +401,7 @@ impl TelemetryInner {
let tracer = tracer_provider.tracer("anwesen"); let tracer = tracer_provider.tracer("anwesen");
tracing::info!( tracing::info!(
endpoint = %env.describe(), endpoint = %endpoint,
slow_request_ms = slow_request.as_millis(), slow_request_ms = slow_request.as_millis(),
"telemetry: OTLP export enabled" "telemetry: OTLP export enabled"
); );
@ -635,208 +489,93 @@ mod tests {
RawTelemetryArgs::default() RawTelemetryArgs::default()
} }
fn generic(endpoint: &str) -> OtelEnv { #[test]
OtelEnv { fn resolve_off_when_nothing_set() {
endpoint: Some(endpoint.into()), let cfg = TelemetryConfig::resolve(raw()).unwrap();
..OtelEnv::default() assert!(cfg.is_none(), "no endpoint/DSN means telemetry off");
}
} }
#[test] #[test]
fn resolve_off_when_no_endpoint_variable_is_set() { fn resolve_generic_endpoint() {
let cfg = TelemetryConfig::resolve(&raw(), OtelEnv::default()).unwrap(); let cfg = TelemetryConfig::resolve(RawTelemetryArgs {
assert!(cfg.is_none(), "no OTEL_ endpoint means telemetry off"); otlp_endpoint: Some("https://collector.example.com".into()),
} otlp_headers: vec!["authorization=Bearer xyz".into()],
slow_request_ms: 500,
#[test] ..raw()
fn resolve_on_for_each_endpoint_variable() { })
let per_signal = |field: fn(&mut OtelEnv)| { .unwrap()
let mut env = OtelEnv::default(); .expect("telemetry on");
field(&mut env); assert_eq!(cfg.endpoint, "https://collector.example.com");
env assert_eq!(
}; cfg.headers,
for env in [ vec![("authorization".to_string(), "Bearer xyz".to_string())]
generic("https://collector.example.com"),
per_signal(|e| {
e.metrics_endpoint = Some("https://collector.example.com/v1/metrics".into());
}),
per_signal(|e| {
e.traces_endpoint = Some("https://collector.example.com/v1/traces".into());
}),
] {
let cfg = TelemetryConfig::resolve(
&RawTelemetryArgs {
slow_request_ms: 250,
..raw()
},
env.clone(),
)
.unwrap_or_else(|e| panic!("{env:?}: {e}"))
.expect("telemetry on");
assert_eq!(cfg.slow_request, Duration::from_millis(250));
}
}
/// The tails the SDK mangles: a query lands in the query string with the
/// signal path behind it, a fragment drops the signal path entirely.
#[test]
fn resolve_rejects_a_query_or_fragment_on_the_generic_endpoint() {
for endpoint in [
"https://api.uptrace.dev?grpc=4317",
"https://api.uptrace.dev/?grpc=4317",
"https://api.uptrace.dev:4318?grpc=4317",
"https://api.uptrace.dev#frag",
] {
let err = TelemetryConfig::resolve(&raw(), generic(endpoint)).unwrap_err();
assert!(err.to_string().contains("trailing"), "{endpoint}: {err}");
}
}
/// A path prefix composes correctly (`/otlp` -> `POST /otlp/v1/metrics`),
/// and the per-signal variables are used verbatim, tail and all.
#[test]
fn resolve_accepts_a_path_prefix_and_per_signal_tails() {
for env in [
generic("https://collector.example.com/otlp"),
generic("https://collector.example.com/"),
OtelEnv {
metrics_endpoint: Some("https://api.uptrace.dev/v1/metrics?grpc=4317".into()),
..OtelEnv::default()
},
] {
TelemetryConfig::resolve(&raw(), env.clone())
.unwrap_or_else(|e| panic!("{env:?}: {e}"))
.expect("telemetry on");
}
}
/// `.with_http()` fixes the transport at compile time, so a protocol this
/// build cannot speak is a dead export with nothing in the log.
#[test]
fn resolve_rejects_a_protocol_this_build_cannot_speak() {
let with_protocol = |field: fn(&mut OtelEnv)| {
let mut env = generic("https://collector.example.com");
field(&mut env);
env
};
for env in [
with_protocol(|e| e.protocol = Some("grpc".into())),
with_protocol(|e| e.protocol = Some("http/json".into())),
with_protocol(|e| e.metrics_protocol = Some("grpc".into())),
with_protocol(|e| e.traces_protocol = Some("grpc".into())),
] {
let err = TelemetryConfig::resolve(&raw(), env.clone())
.unwrap_err()
.to_string();
assert!(err.contains("http/protobuf"), "{env:?}: {err}");
}
}
/// The default protocol is the one this build speaks, so naming it
/// explicitly is not an error.
#[test]
fn resolve_accepts_the_supported_protocol_spelled_out() {
let env = OtelEnv {
protocol: Some("http/protobuf".into()),
metrics_protocol: Some("http/protobuf".into()),
..generic("https://collector.example.com")
};
TelemetryConfig::resolve(&raw(), env)
.unwrap()
.expect("telemetry on");
}
/// Contract point 6: `service.name` stays anwesen's default only while
/// the environment supplies none. A builder attribute wins over the SDK's
/// detectors, so the default has to step aside for the override to work.
#[test]
fn service_identity_defaults_step_aside_for_the_environment() {
let keys = |attrs: &[KeyValue]| {
attrs
.iter()
.map(|kv| kv.key.as_str().to_string())
.collect::<Vec<_>>()
};
let plain = default_resource_attrs(None, None);
assert_eq!(keys(&plain), [SERVICE_NAME, SERVICE_VERSION]);
let named = default_resource_attrs(Some("svcname-override"), None);
assert_eq!(keys(&named), [SERVICE_VERSION]);
let attrs = default_resource_attrs(
None,
Some("deployment.environment=prod,service.name=attrs-override"),
); );
assert_eq!(keys(&attrs), [SERVICE_VERSION]); assert_eq!(cfg.slow_request, Duration::from_millis(500));
let versioned = default_resource_attrs(None, Some("service.version=9.9.9"));
assert_eq!(keys(&versioned), [SERVICE_NAME]);
// An empty OTEL_SERVICE_NAME supplies nothing; the SDK ignores it too.
let empty = default_resource_attrs(Some(" "), None);
assert_eq!(keys(&empty), [SERVICE_NAME, SERVICE_VERSION]);
} }
#[test] #[test]
fn resolve_rejects_the_removed_uptrace_dsn() { fn resolve_uptrace_dsn_splits_endpoint_and_header() {
let err = TelemetryConfig::resolve( let cfg = TelemetryConfig::resolve(RawTelemetryArgs {
&RawTelemetryArgs { uptrace_dsn: Some("https://SECRET_TOKEN@api.uptrace.dev".into()),
uptrace_dsn: Some("https://tok@api.uptrace.dev".into()), slow_request_ms: 250,
..raw() ..raw()
}, })
OtelEnv::default(), .unwrap()
) .expect("telemetry on");
.unwrap_err() assert_eq!(cfg.endpoint, "https://api.uptrace.dev");
.to_string(); assert_eq!(
assert!(err.contains("--uptrace-dsn"), "{err}"); cfg.headers,
assert!(err.contains("OTEL_EXPORTER_OTLP_HEADERS"), "{err}"); vec![(
"uptrace-dsn".to_string(),
"https://SECRET_TOKEN@api.uptrace.dev".to_string()
)]
);
assert_eq!(cfg.slow_request, Duration::from_millis(250));
} }
#[test] #[test]
fn resolve_rejects_the_removed_otlp_endpoint() { fn resolve_dsn_header_leads_explicit_headers() {
let err = TelemetryConfig::resolve( let cfg = TelemetryConfig::resolve(RawTelemetryArgs {
&RawTelemetryArgs { uptrace_dsn: Some("https://tok@api.uptrace.dev".into()),
otlp_endpoint: Some("https://collector.example.com".into()), otlp_headers: vec!["x-extra=1".into()],
..raw() ..raw()
}, })
OtelEnv::default(), .unwrap()
) .expect("telemetry on");
.unwrap_err() assert_eq!(cfg.headers[0].0, "uptrace-dsn");
.to_string(); assert_eq!(cfg.headers[1], ("x-extra".to_string(), "1".to_string()));
assert!(err.contains("--otlp-endpoint"), "{err}");
assert!(err.contains("OTEL_EXPORTER_OTLP_ENDPOINT"), "{err}");
} }
#[test] #[test]
fn resolve_rejects_the_removed_otlp_header() { fn resolve_rejects_both_endpoint_sources() {
let err = TelemetryConfig::resolve( let err = TelemetryConfig::resolve(RawTelemetryArgs {
&RawTelemetryArgs { uptrace_dsn: Some("https://tok@api.uptrace.dev".into()),
otlp_headers: vec!["authorization=Bearer xyz".into()], otlp_endpoint: Some("https://collector.example.com".into()),
..raw() ..raw()
}, })
OtelEnv::default(),
)
.unwrap_err()
.to_string();
assert!(err.contains("--otlp-header"), "{err}");
assert!(err.contains("OTEL_EXPORTER_OTLP_HEADERS"), "{err}");
}
/// A removed option fails even with a correct `OTEL_` endpoint alongside
/// it: the operator's intent is in the flag, and half-applied config is
/// the silent failure this issue removes.
#[test]
fn resolve_rejects_a_removed_option_alongside_a_good_endpoint() {
let err = TelemetryConfig::resolve(
&RawTelemetryArgs {
uptrace_dsn: Some("https://tok@api.uptrace.dev".into()),
..raw()
},
generic("https://api.uptrace.dev"),
)
.unwrap_err(); .unwrap_err();
assert!(err.to_string().contains("removed"), "{err}"); assert!(err.to_string().contains("mutually exclusive"));
}
#[test]
fn resolve_rejects_malformed_header() {
let err = TelemetryConfig::resolve(RawTelemetryArgs {
otlp_endpoint: Some("https://collector.example.com".into()),
otlp_headers: vec!["no-equals-sign".into()],
..raw()
})
.unwrap_err();
assert!(err.to_string().contains("key=value"));
}
#[test]
fn resolve_rejects_dsn_without_scheme() {
let err = TelemetryConfig::resolve(RawTelemetryArgs {
uptrace_dsn: Some("tok@api.uptrace.dev".into()),
..raw()
})
.unwrap_err();
assert!(err.to_string().contains("scheme"));
} }
#[test] #[test]

4
src/watcher.rs
View file

@ -4,7 +4,7 @@
//! [`notify::RecommendedWatcher`] over the vault root. Each native event is //! [`notify::RecommendedWatcher`] over the vault root. Each native event is
//! pushed into a Tokio channel; [`run_debouncer`] drains the channel, //! pushed into a Tokio channel; [`run_debouncer`] drains the channel,
//! classifies events into [`WatchAction`]s, coalesces a 100 ms window's //! classifies events into [`WatchAction`]s, coalesces a 100 ms window's
//! worth into one [`WatchBatch`], and casts the batch to the `index_writer` //! worth into one [`Batch`], and casts the batch to the `index_writer`
//! named process for a single [`crate::store::NoteStore`] write. //! named process for a single [`crate::store::NoteStore`] write.
//! //!
//! See [[ADR-003 Filesystem Change Tracking]] for the event model. //! See [[ADR-003 Filesystem Change Tracking]] for the event model.
@ -27,7 +27,7 @@ use crate::vault;
/// One path-scoped action derived from a native filesystem event. Always /// One path-scoped action derived from a native filesystem event. Always
/// carries a vault-relative, forward-slash-normalized path string -- the /// carries a vault-relative, forward-slash-normalized path string -- the
/// same form [`vault::Note::path`] uses. /// same form [`vault::Note.path`] uses.
/// ///
/// The `*Tree` variants carry a directory instead of a note. Native events /// The `*Tree` variants carry a directory instead of a note. Native events
/// name only the directory when one is created, removed, or renamed; the /// name only the directory when one is created, removed, or renamed; the

14
tests/run-hurl.sh
View file

@ -54,20 +54,6 @@ if [[ $ready -ne 1 ]]; then
exit 1 exit 1
fi fi
# ANW-43: the offline `query` subcommand must return the document the endpoint
# returns. Both surfaces read the same fixture vault, so the two outputs are
# compared byte for byte before the contract suite runs.
for q in "" "tags=project" "__anw-path=Projects&__anw-limit=1" "status__exists=true"; do
cli="$("$BIN" query --vault "$VAULT" --query "$q" --log-level error)"
http="$(curl -sf "http://$HOST:$PORT/query?$q")"
if [[ "$cli" != "$http" ]]; then
echo "anwesen query and GET /query disagree for query '$q'" >&2
echo " cli: $cli" >&2
echo " http: $http" >&2
exit 1
fi
done
# Run every *.hurl. Glob into an array so we can fail loud if there are none. # Run every *.hurl. Glob into an array so we can fail loud if there are none.
shopt -s globstar nullglob shopt -s globstar nullglob
files=("$ROOT"/tests/hurl/**/*.hurl) files=("$ROOT"/tests/hurl/**/*.hurl)

33
tests/serve_start_failure.rs
View file

@ -1,33 +0,0 @@
//! `anwesen serve` must exit nonzero when the supervisor tree fails to start
//! ([ANW-45](https://crvrs.youtrack.cloud/issue/ANW-45)). Hydra's
//! `Application::run` logs the failure and returns normally, so without an
//! explicit check the process exits 0 and systemd's `Restart=on-failure`
//! never retries -- the vault stayed unreachable for 12 hours on ap.
//!
//! The forced failure is a taken bind address: `http_server` binds eagerly in
//! its child spec, so the address-in-use error fails the whole start.
use std::net::TcpListener;
use std::process::Command;
#[test]
fn serve_exits_nonzero_when_the_tree_fails_to_start() {
let vault = tempfile::tempdir().expect("tempdir");
// Hold the port for the lifetime of the child so its bind cannot succeed.
let held = TcpListener::bind("127.0.0.1:0").expect("bind probe port");
let addr = held.local_addr().expect("probe addr");
let status = Command::new(env!("CARGO_BIN_EXE_anwesen"))
.arg("serve")
.arg("--vault")
.arg(vault.path())
.arg("--bind")
.arg(addr.to_string())
.status()
.expect("spawn anwesen serve");
assert!(
!status.success(),
"serve exited {status} after a failed supervisor start; systemd reads that as a clean exit"
);
}