Compare commits

...

10 commits
v0.1.0 ... main

Author SHA1 Message Date
e1eada39ab ANW-47 Release: bump version to 0.4.0
Some checks failed
release / aarch64-linux (push) Has been cancelled
release / aarch64-macos (push) Has been cancelled
release / x86_64-linux (push) Has been cancelled
release / publish (push) Has been cancelled
2026-07-31 16:22:25 +03:00
6e07bff08a ANW-43 CLI: add a query subcommand writing the /query JSON offline
The README claims the query engine is available offline, but only the
merge half was: no subcommand produced the path / frontmatter /
last_modified / etag / size projection that GET /query returns.

`anwesen query` takes the same --vault and --query flags as merge and
writes that document to stdout, compact, one line. src/merge.rs becomes
src/oneshot.rs and holds both: one parse-and-load path, so the two
subcommands cannot drift in grammar or in how strictly they read a
vault. cli.rs shares one VaultQueryArgs between them for the same
reason. rfc3339_z moves from http.rs to query.rs, next to the projection
whose timestamp dialect it is.

Assumed a trailing newline on stdout is wanted here even though merge
writes none: the JSON body is one line for jq and shell pipelines, and
the newline terminates it rather than joining the document. Flag if
wrong.

The hurl harness now compares `anwesen query` against GET /query on the
fixture vault for four queries before running the suite, so a drift
between the two surfaces fails CI.
2026-07-31 15:59:18 +03:00
30bca24b5c ANW-45 Serve: exit nonzero when the supervisor tree fails to start
Hydra's Application::run logs a start failure and returns normally, so
serve exited 0 after the tree never came up. systemd read that as a clean
start and Restart=on-failure never retried; anwesen was unreachable on ap
for 12 hours with NRestarts=0.

Anwesen carries a started flag that start() sets once the supervisor is
up, and main exits 1 when it is still false after run() returns.

Assumed the flag is the only seam available: run() consumes self and
returns (), so the Err is not observable at the call site. Flag if a newer
hydra exposes the result.
2026-07-29 22:22:24 +03:00
d6261bd5f9 ANW-44 Release: bump version to 0.3.0
Some checks failed
release / aarch64-linux (push) Has been cancelled
release / aarch64-macos (push) Has been cancelled
release / x86_64-linux (push) Has been cancelled
release / publish (push) Has been cancelled
2026-07-27 00:21:53 +03:00
dc113b8249 ANW-42 Telemetry: take export config from OTEL_ environment variables only
Delete the uptrace DSN parser and the {endpoint}/v1/* concatenation that
produced the dead export, and build the OTLP exporters with no endpoint,
headers, or protocol so the SDK reads OTEL_EXPORTER_OTLP_* itself.

The SDK concatenates as naively as we did: measured against a local sink,
a ?query base sends POST /?query/v1/metrics and a #fragment base sends
POST /. So one check stays -- a query or fragment on
OTEL_EXPORTER_OTLP_ENDPOINT fails at startup. Per-signal variables are
used verbatim and need none.

A protocol variable is the second startup check. The exporter picks its
transport at build time and this binary ships OTLP/HTTP alone, so
OTEL_EXPORTER_OTLP_PROTOCOL=grpc kept exporting over HTTP with nothing in
the log -- the same silence this change removes.

service.name stays a default rather than policy: an attribute set on the
resource builder wins over the SDK detectors that read OTEL_SERVICE_NAME
and OTEL_RESOURCE_ATTRIBUTES, so anwesen supplies its own only for keys
the environment leaves alone.

Telemetry-off is our own check on the endpoint variables: with none set
the exporter would still build and aim at the SDK default localhost:4318.

--uptrace-dsn, --otlp-endpoint, and --otlp-header stay parsed but hidden,
so a deployment upgrading with them set fails naming the OTEL_
replacement instead of going quiet.

Assumed hidden clap stubs are the right migration shape; clap rejecting
the flags as unknown would say nothing about the replacement. Flag if a
plain unknown-argument error is wanted instead.

Assumed rejecting http/json alongside grpc is right: neither has a
transport in this build. Flag if a build with both features is wanted
instead.
2026-07-27 00:06:36 +03:00
8894ea0c7b ANW-41 Docs: fix four broken rustdoc intra-doc links
cargo doc --no-deps is warning-free. No behaviour change.
2026-07-24 23:41:37 +03:00
134b1d44b3 ANW-39 Release: bump version to 0.2.0
Some checks failed
release / aarch64-linux (push) Has been cancelled
release / aarch64-macos (push) Has been cancelled
release / x86_64-linux (push) Has been cancelled
release / publish (push) Has been cancelled
2026-07-24 21:42:18 +03:00
bbb052fdc7 ANW-40 Watcher: emit both deletes for a vanished .md path
A directory can be named *.md too, and a vanished path cannot be probed,
so the suffix test alone stranded its notes in the index. Emit the note
delete and the prefix delete together; the two key sets are disjoint.
2026-07-24 21:32:51 +03:00
275c6ecfa0 ANW-36 Watcher: drop index entries when note files are deleted 2026-07-24 21:19:13 +03:00
23ce90e103 ANW-37 Telemetry: request metrics + selective traces over OTLP
Optional OTLP export of request-level metrics (request and response-byte
counters keyed by route/status/conditional-GET, a duration histogram out
to 600s) plus server spans for slow or 5xx requests, nested under the
incoming W3C trace context. Off unless an OTLP endpoint or uptrace DSN is
configured; a config-less server is byte-for-byte unchanged.
2026-07-24 21:08:07 +03:00
18 changed files with 3339 additions and 341 deletions

1199
Cargo.lock generated

File diff suppressed because it is too large Load diff

View file

@ -1,7 +1,7 @@
[package]
name = "anwesen"
description = "Read-only HTTP daemon over a markdown vault, querying YAML frontmatter."
version = "0.1.0"
version = "0.4.0"
edition = "2024"
rust-version = "1.95"
license = "BSD-3-Clause"
@ -17,8 +17,17 @@ axum = "0.8"
blake3 = "1"
chrono = { version = "0.4", default-features = false, features = ["std", "serde", "clock"] }
clap = { version = "4", features = ["derive", "env"] }
http-body = "1"
hydra = "0.1"
notify = "8"
opentelemetry = "0.32"
# `reqwest-rustls` is required alongside `reqwest-blocking-client`: the
# default otlp features wire reqwest with no TLS backend at all, so HTTPS to
# the collector would fail. rustls (aws-lc-rs + OS trust store) keeps the
# release binary free of a system OpenSSL dependency.
opentelemetry-otlp = { version = "0.32", default-features = false, features = ["trace", "metrics", "http-proto", "reqwest-blocking-client", "reqwest-rustls"] }
opentelemetry-semantic-conventions = "0.32"
opentelemetry_sdk = { version = "0.32", features = ["metrics", "trace"] }
regex = "1"
serde = { version = "1", features = ["derive"] }
serde_json = "1"

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 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)).
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)).
## Quick start
@ -52,29 +52,86 @@ 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
```
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
```
anwesen serve --vault <path> [--bind <addr:port>] [--log-level <level>]
[--otlp-slow-request-ms <n>]
anwesen doctor --vault <path>
anwesen merge --vault <path> --query <query-string>
anwesen query --vault <path> --query <query-string>
anwesen version
```
| Flag | Env var | Default | Meaning |
| ------------------------ | ------------------- | ---------------------- | ----------------------------------------------------------------------- |
| `--vault <path>` | `ANWESEN_VAULT` | _required_ | Path to the vault root. |
| `--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`. |
| `--query <query-string>` | -- | _required for `merge`_ | A `/query` query string: frontmatter predicates plus `__anw-` controls. |
| Flag | Env var | Default | Meaning |
| --------------------------- | ------------------------------ | ---------------------- | ----------------------------------------------------------------------------- |
| `--vault <path>` | `ANWESEN_VAULT` | _required_ | Path to the vault root. |
| `--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`. |
| `--query <query-string>` | `ANWESEN_QUERY` | empty (match all) | A `/query` query string: frontmatter predicates plus `__anw-` controls. `merge` and `query` only. |
| `--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 over env vars.
Every flag has a matching `ANWESEN_<UPPER>` environment variable. CLI flags win
over env vars.
`--bind` and `--otlp-slow-request-ms` apply to `serve` only. Where telemetry is
exported is configured entirely through the standard `OTEL_` variables below.
- **`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.
- **`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.
## 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
All endpoints are `GET` and return JSON unless noted.
@ -124,7 +181,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-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>`.
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)).
#### Markdown-merge mode
@ -152,7 +209,11 @@ Returns vault path, note count, last index/event timestamps, watcher state, an i
## Local generation
`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:
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.
### `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'
@ -162,6 +223,24 @@ 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.
### `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
- **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.

View file

@ -40,6 +40,7 @@ use tokio::task::JoinHandle;
use crate::health::HealthState;
use crate::http::{self as http_layer, HttpState};
use crate::store::NoteStore;
use crate::telemetry::Telemetry;
use crate::vault::{self, Note};
use crate::watcher::run_debouncer;
@ -117,17 +118,28 @@ pub struct Anwesen {
pub store: Arc<NoteStore>,
/// Shared mutable health surface consumed by `/health` ([ANW-8]).
pub health: Arc<HealthState>,
/// Request-level telemetry handle ([ANW-37]). `None` disables export and
/// the request middleware entirely.
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 {
#[must_use]
pub fn new(vault: PathBuf, bind: SocketAddr) -> Self {
pub fn new(vault: PathBuf, bind: SocketAddr, telemetry: Option<Arc<Telemetry>>) -> Self {
Self {
vault,
bind,
counters: RestartCounters::new(),
store: NoteStore::new(),
health: HealthState::new(),
telemetry,
started: Arc::new(AtomicBool::new(false)),
}
}
}
@ -177,14 +189,17 @@ impl Application for Anwesen {
store: self.store.clone(),
health: self.health.clone(),
vault: self.vault.clone(),
telemetry: self.telemetry.clone(),
}
.child_spec(),
];
Supervisor::with_children(children)
let pid = Supervisor::with_children(children)
.strategy(SupervisionStrategy::OneForOne)
.start_link(SupervisorOptions::new().name("anwesen_root"))
.await
.await?;
self.started.store(true, Ordering::Release);
Ok(pid)
}
}
@ -436,6 +451,14 @@ pub enum IndexWriterMessage {
pub struct IndexBatch {
pub upserts: Vec<Note>,
pub deletes: Vec<String>,
/// Directories whose indexed notes all drop before the upserts land. A
/// removed or renamed directory produces no per-file event, so the
/// prefix is the only handle on those notes. A directory that was
/// walked is listed here too: the index under that prefix has to match
/// the walk, not keep what an earlier directory of the same name left
/// behind [ANW-36].
#[serde(default)]
pub delete_prefixes: Vec<String>,
}
#[derive(Clone)]
@ -492,8 +515,15 @@ impl GenServer for IndexWriterState {
}
IndexWriterMessage::Batch(batch) => {
let (u, d) = (batch.upserts.len(), batch.deletes.len());
self.store.apply_batch(batch.upserts, &batch.deletes);
tracing::info!(upserts = u, deletes = d, "index_writer: batch applied");
let dropped =
self.store
.apply_batch(batch.upserts, &batch.deletes, &batch.delete_prefixes);
tracing::info!(
upserts = u,
deletes = d,
dropped,
"index_writer: batch applied"
);
}
IndexWriterMessage::Upsert(note) => {
let path = note.path.clone();
@ -535,6 +565,7 @@ pub struct HttpServer {
store: Arc<NoteStore>,
health: Arc<HealthState>,
vault: PathBuf,
telemetry: Option<Arc<Telemetry>>,
}
impl HttpServer {
@ -544,12 +575,14 @@ impl HttpServer {
let store = self.store.clone();
let health = self.health.clone();
let vault = self.vault.clone();
let telemetry = self.telemetry.clone();
ChildSpec::new(HTTP_SERVER_NAME).start(move || {
let bind = bind;
let counters = counters.clone();
let store = store.clone();
let health = health.clone();
let vault = vault.clone();
let telemetry = telemetry.clone();
async move {
// Bind eagerly so the supervisor sees `bind: <addr>: ...`
// as the start error rather than a successful spawn that
@ -558,12 +591,15 @@ impl HttpServer {
.await
.map_err(|e| ExitReason::from(format!("http_server: bind {bind}: {e}")))?;
let restart = counters.http_server.record_init();
let router = http_layer::router(HttpState {
store,
health,
restart_counters: counters,
vault,
});
let router = http_layer::router(
HttpState {
store,
health,
restart_counters: counters,
vault,
},
telemetry,
);
let pid = Process::spawn_link(async move {
Process::set_flags(ProcessFlags::TRAP_EXIT);
tracing::info!(restart, bind = %bind, "http_server: init");

View file

@ -4,14 +4,27 @@
//!
//! ```text
//! anwesen serve --vault <path> [--bind <addr:port>] [--log-level <level>]
//! [--otlp-slow-request-ms <n>]
//! anwesen doctor --vault <path> [--log-level <level>]
//! anwesen merge --vault <path> [--query <string>] [--log-level <level>]
//! anwesen query --vault <path> [--query <string>] [--log-level <level>]
//! anwesen version
//! ```
//!
//! `--bind` is `serve`-only; `doctor` and `merge` do not bind a port;
//! `version` takes no flags. Each flag has a matching `ANWESEN_<UPPER>`
//! environment variable and CLI wins over env per the manual.
//! `merge` and `query` differ only in what they write -- the merged markdown
//! 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
//! flag has a matching `ANWESEN_<UPPER>` environment variable and CLI wins
//! over env per the manual.
//!
//! 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::path::PathBuf;
@ -33,7 +46,11 @@ pub enum Command {
Doctor(DoctorArgs),
/// Walk the vault once, evaluate the query, and write the merged markdown
/// document to stdout. One-shot; no server. Read-only.
Merge(MergeArgs),
Merge(VaultQueryArgs),
/// 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.
Version,
}
@ -51,6 +68,31 @@ pub struct ServeArgs {
/// Log verbosity.
#[arg(long, env = "ANWESEN_LOG_LEVEL", default_value = "info")]
pub log_level: LogLevel,
/// Removed (ANW-42): use `OTEL_EXPORTER_OTLP_ENDPOINT` plus
/// `OTEL_EXPORTER_OTLP_HEADERS=uptrace-dsn=<dsn>`. Still accepted so
/// startup fails with that message rather than exporting nowhere.
#[arg(long, env = "ANWESEN_UPTRACE_DSN", hide = true)]
pub uptrace_dsn: Option<String>,
/// Removed (ANW-42): use `OTEL_EXPORTER_OTLP_ENDPOINT`.
#[arg(long, env = "ANWESEN_OTLP_ENDPOINT", hide = true)]
pub otlp_endpoint: Option<String>,
/// Removed (ANW-42): use `OTEL_EXPORTER_OTLP_HEADERS`.
#[arg(
long = "otlp-header",
env = "ANWESEN_OTLP_HEADERS",
value_delimiter = ',',
hide = true
)]
pub otlp_headers: Vec<String>,
/// Requests at or over this duration in milliseconds, or answering a
/// 5xx, are additionally recorded as OTLP server spans; every other
/// request stays metrics-only.
#[arg(long, env = "ANWESEN_OTLP_SLOW_REQUEST_MS", default_value = "500")]
pub otlp_slow_request_ms: u64,
}
#[derive(Debug, clap::Args)]
@ -64,8 +106,9 @@ pub struct DoctorArgs {
pub log_level: LogLevel,
}
/// Flags shared by the two one-shot subcommands, `merge` and `query`.
#[derive(Debug, clap::Args)]
pub struct MergeArgs {
pub struct VaultQueryArgs {
/// Path to the vault root.
#[arg(long, env = "ANWESEN_VAULT")]
pub vault: PathBuf,
@ -73,12 +116,12 @@ pub struct MergeArgs {
/// Query in the `/query` query-string grammar, for example
/// `tags=anwesen&__anw-kind=skill&__anw-order=order`. The `__anw-kind`
/// homogeneity guard and `__anw-order` fragment ordering ride inside this
/// string -- there are no separate flags. Empty merges every note under
/// the vault root.
/// string -- there are no separate flags. `__anw-kind` and `__anw-order`
/// apply to `merge` only. Empty matches every note under the vault root.
#[arg(long, env = "ANWESEN_QUERY", default_value = "")]
pub query: String,
/// Log verbosity. Logs go to stderr; the merged document goes to stdout.
/// Log verbosity. Logs go to stderr; the document goes to stdout.
#[arg(long, env = "ANWESEN_LOG_LEVEL", default_value = "info")]
pub log_level: LogLevel,
}
@ -140,6 +183,34 @@ 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]
fn serve_rejects_malformed_bind_at_parse_time() {
let err = parse(&["serve", "--vault", "/tmp/v", "--bind", "not-an-addr"]).unwrap_err();
@ -210,6 +281,47 @@ mod tests {
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]
fn version_takes_no_flags() {
assert!(matches!(

View file

@ -46,7 +46,7 @@ pub struct DriftShape {
/// `"bool"`, `"number"`, `"string"`, `"date"`, `"list"`, `"mapping"`.
pub shape: &'static str,
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.
pub samples: Vec<String>,
}

View file

@ -12,16 +12,18 @@
//! `/health` ([ANW-8]) onto the same [`Router`].
use std::sync::Arc;
use std::time::{Instant, SystemTime};
use axum::Router;
use axum::body::Body;
use axum::extract::{OriginalUri, Path as AxumPath, Request, State};
use axum::http::header::{ACCEPT, CONTENT_TYPE, ETAG, IF_NONE_MATCH};
use axum::http::header::{ACCEPT, CONTENT_LENGTH, CONTENT_TYPE, ETAG, IF_NONE_MATCH};
use axum::http::{HeaderMap, HeaderValue, StatusCode};
use axum::middleware::{Next, from_fn};
use axum::middleware::{Next, from_fn, from_fn_with_state};
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use chrono::{DateTime, SecondsFormat, Utc};
use chrono::{DateTime, Utc};
use http_body::Body as _;
use hydra::Process;
use serde::Serialize;
use std::collections::BTreeMap;
@ -30,16 +32,11 @@ use std::path::PathBuf;
use crate::app::RestartCounters;
use crate::health::HealthState;
use crate::query::rfc3339_z;
use crate::store::NoteStore;
use crate::telemetry::{self, Telemetry, TraceHeaders};
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.
#[derive(Clone)]
pub struct HttpState {
@ -49,18 +46,72 @@ pub struct HttpState {
pub vault: PathBuf,
}
pub fn router(state: HttpState) -> Router {
pub fn router(state: HttpState, telemetry: Option<Arc<Telemetry>>) -> Router {
// `/notes/{*path}` is greedy and includes any trailing slash; one
// handler dispatches read-one vs folder-listing on that suffix. The
// root listing (`/notes/`) needs its own route since the wildcard
// requires at least one character.
Router::new()
let router = Router::new()
.route("/notes/", get(list_root_folder))
.route("/notes/{*path}", get(get_notes))
.route("/query", get(get_query))
.route("/health", get(get_health))
.layer(from_fn(process_wrap))
.with_state(state)
.layer(from_fn(process_wrap));
// Telemetry is the outermost layer so it observes the final response,
// including the `500` that `process_wrap` synthesizes on a handler
// panic. Installed only when export is configured; a config-less server
// never runs this layer and is byte-for-byte as before ([ANW-37]).
let router = match telemetry {
Some(tel) => router.layer(from_fn_with_state(tel, telemetry_wrap)),
None => router,
};
router.with_state(state)
}
/// Outermost middleware: time the request, classify its route, and hand the
/// completed observation to [`Telemetry`]. Trace-context headers are captured
/// up front (the handler consumes the request); the parent context is only
/// extracted later if the request turns out slow enough to span.
async fn telemetry_wrap(
State(tel): State<Arc<Telemetry>>,
request: Request,
next: Next,
) -> Response {
let start_instant = Instant::now();
let start_wall = SystemTime::now();
let route = telemetry::classify_route(request.uri().path());
let method = request.method().clone();
let if_none_match_present = request.headers().contains_key(IF_NONE_MATCH);
let trace = TraceHeaders::from_headers(request.headers());
let response = next.run(request).await;
let status = response.status().as_u16();
// In-memory handler bodies carry an exact size hint; fall back to a
// Content-Length header, then to 0, so a streaming body never panics.
let body_bytes = response
.body()
.size_hint()
.exact()
.or_else(|| content_length(response.headers()))
.unwrap_or(0);
let duration = start_instant.elapsed();
tel.finish(
route,
method.as_str(),
status,
if_none_match_present,
body_bytes,
duration,
start_wall,
&trace,
);
response
}
fn content_length(headers: &HeaderMap) -> Option<u64> {
headers.get(CONTENT_LENGTH)?.to_str().ok()?.parse().ok()
}
/// Per [ANW-25](https://crvrs.youtrack.cloud/issue/ANW-25): run each request
@ -460,12 +511,15 @@ mod tests {
fn router_with(notes: Vec<Note>) -> (Router, Arc<NoteStore>) {
let store = NoteStore::new();
store.replace(notes);
let r = router(HttpState {
store: store.clone(),
health: crate::health::HealthState::new(),
restart_counters: crate::app::RestartCounters::new(),
vault: PathBuf::from("/test/vault"),
});
let r = router(
HttpState {
store: store.clone(),
health: crate::health::HealthState::new(),
restart_counters: crate::app::RestartCounters::new(),
vault: PathBuf::from("/test/vault"),
},
None,
);
(r, store)
}
@ -757,4 +811,51 @@ mod tests {
assert!(resolve_path("").is_err());
assert!(resolve_path("a/%ZZ.md").is_err());
}
/// With telemetry installed, a normal request is answered byte-for-byte
/// as without it. The test process sets no `OTEL_EXPORTER_OTLP_*`
/// variables, so the exporter aims at the SDK default and fails in the
/// background without ever touching the response path.
#[tokio::test]
async fn telemetry_layer_does_not_alter_responses() {
use crate::telemetry::{self, OtelEnv, RawTelemetryArgs, TelemetryConfig};
let cfg = TelemetryConfig::resolve(
&RawTelemetryArgs {
slow_request_ms: 500,
..Default::default()
},
OtelEnv {
endpoint: Some("http://127.0.0.1:9".into()),
..OtelEnv::default()
},
)
.unwrap()
.expect("telemetry on");
let tel = Arc::new(telemetry::init(cfg).expect("telemetry init"));
let n = note("a.md", "body");
let expected_etag = n.etag.clone();
let store = NoteStore::new();
store.replace(vec![n]);
let r = router(
HttpState {
store,
health: crate::health::HealthState::new(),
restart_counters: crate::app::RestartCounters::new(),
vault: PathBuf::from("/test/vault"),
},
Some(tel.clone()),
);
let req = Request::get("/notes/a.md").body(Body::empty()).unwrap();
let resp = send(r, req).await;
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(resp.headers().get(ETAG).unwrap(), expected_etag.as_str());
let bytes = to_bytes(resp.into_body(), 64 * 1024).await.unwrap();
let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(v["path"], "a.md");
assert_eq!(v["body"], "body");
tel.shutdown();
}
}

View file

@ -10,8 +10,9 @@ pub mod app;
pub mod doctor;
pub mod health;
pub mod http;
pub mod merge;
pub mod oneshot;
pub mod query;
pub mod store;
pub mod telemetry;
pub mod vault;
pub mod watcher;

View file

@ -1,13 +1,17 @@
//! Anwesen: read-only HTTP daemon over a markdown vault.
//!
//! This module wires the CLI to the `serve`, `doctor`, `merge`, and
//! This module wires the CLI to the `serve`, `doctor`, `merge`, `query`, and
//! `version` subcommands.
mod cli;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use anwesen::app::Anwesen;
use anwesen::doctor;
use anwesen::merge;
use anwesen::oneshot;
use anwesen::telemetry::{self, OtelEnv, RawTelemetryArgs, TelemetryConfig};
use anyhow::Result;
use clap::Parser;
use hydra::Application;
@ -15,21 +19,50 @@ use hydra::Application;
use crate::cli::{Cli, Command};
// Result is retained at the binary boundary per [[ADR-001 Language and
// Foundation Libraries]] (anyhow at main); current stubs do not yet
// surface errors.
#[allow(clippy::unnecessary_wraps)]
// Foundation Libraries]] (anyhow at main); telemetry config resolution and
// exporter setup surface startup errors through it.
fn main() -> Result<()> {
let cli = Cli::parse();
match cli.command {
Command::Serve(args) => {
init_logging(args.log_level);
// Resolve telemetry config before the supervisor starts; no
// OTEL_EXPORTER_OTLP_* endpoint leaves it `None` (export off,
// no middleware). A removed flag is a startup error (ANW-42).
let telemetry = match TelemetryConfig::resolve(
&RawTelemetryArgs {
uptrace_dsn: args.uptrace_dsn,
otlp_endpoint: args.otlp_endpoint,
otlp_headers: args.otlp_headers,
slow_request_ms: args.otlp_slow_request_ms,
},
OtelEnv::from_env(),
)? {
Some(cfg) => Some(Arc::new(telemetry::init(cfg)?)),
None => None,
};
tracing::info!(
vault = %args.vault.display(),
bind = %args.bind,
telemetry = telemetry.is_some(),
"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).
Anwesen::new(args.vault, args.bind).run();
app.run();
// Flush and shut down exporters after the server loop returns.
if let Some(telemetry) = telemetry {
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) => {
init_logging(args.log_level);
@ -41,13 +74,26 @@ fn main() -> Result<()> {
}
Command::Merge(args) => {
init_logging(args.log_level);
match merge::run(&args.vault, &args.query) {
match oneshot::merge(&args.vault, &args.query) {
// `print!`, not `println!`: the merged document is byte-stable
// and byte-identical to the HTTP merge body, which carries no
// trailing newline. An empty match set prints nothing, exit 0.
Ok(doc) => print!("{doc}"),
Err(e) => {
eprint!("{}", e.render());
eprint!("{}", e.render("merge"));
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);
}
}

View file

@ -1,205 +0,0 @@
//! 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 Normal file
View file

@ -0,0 +1,354 @@
//! 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);
}
}

View file

@ -11,14 +11,14 @@
//! `__in` / `__all` are comma-separated; unknown operators are `400`.
//!
//! Predicates are evaluated by iterating the in-memory [`NoteStore`] and
//! applying each [`Predicate::matches`] in turn. At the documented scale
//! applying each [`Predicate`] in turn. At the documented scale
//! (low-thousands-of-notes vaults) this is sub-millisecond; see
//! [[ADR-009 Reverse ADR-002 In-Memory Evaluation No Tantivy]] for the
//! call to keep evaluation in-memory rather than carrying a Tantivy index.
use std::collections::BTreeMap;
use chrono::{DateTime, NaiveDate};
use chrono::{DateTime, NaiveDate, SecondsFormat, Utc};
use regex::Regex;
use serde::Serialize;
use serde_json::Value as JsonValue;
@ -280,6 +280,16 @@ 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
/// elided -- consumers fetch bodies via `/notes/<path>` if needed.
#[derive(Debug, Serialize)]

View file

@ -40,19 +40,45 @@ impl NoteStore {
}
}
/// Apply one debounce-window's worth of changes. Mirrors the order in
/// [`crate::index::NoteIndex::apply_batch`]: deletes first, then upserts.
/// Apply one debounce-window's worth of changes: deletes first (paths,
/// then whole directories), then upserts. A "delete-then-upsert"
/// sequence on one path is therefore unambiguous, and a directory that
/// was removed and recreated inside one window keeps only what the walk
/// found.
///
/// Each entry of `delete_prefixes` is a vault-relative directory; every
/// note under it drops [ANW-36]. Returns the number of notes dropped
/// that way.
///
/// # Panics
/// Panics if the inner `RwLock` has been poisoned.
pub fn apply_batch(&self, upserts: Vec<Note>, deletes: &[String]) {
pub fn apply_batch(
&self,
upserts: Vec<Note>,
deletes: &[String],
delete_prefixes: &[String],
) -> usize {
let mut guard = self.inner.write().expect("note_store: write lock poisoned");
for path in deletes {
guard.remove(path);
}
let mut dropped = 0;
for dir in delete_prefixes {
let prefix = format!("{}/", dir.trim_end_matches('/'));
let doomed: Vec<String> = guard
.range(prefix.clone()..)
.take_while(|(p, _)| p.starts_with(&prefix))
.map(|(p, _)| p.clone())
.collect();
dropped += doomed.len();
for path in doomed {
guard.remove(&path);
}
}
for note in upserts {
guard.insert(note.path.clone(), note);
}
dropped
}
/// # Panics
@ -143,13 +169,65 @@ mod tests {
fn apply_batch_deletes_then_upserts() {
let s = NoteStore::new();
s.replace(vec![note("a.md"), note("b.md")]);
s.apply_batch(vec![note("c.md")], &["a.md".to_string()]);
s.apply_batch(vec![note("c.md")], &["a.md".to_string()], &[]);
assert_eq!(s.len(), 2);
assert!(s.get("a.md").is_none());
assert!(s.get("b.md").is_some());
assert!(s.get("c.md").is_some());
}
#[test]
fn apply_batch_prefix_drops_the_whole_subtree() {
let s = NoteStore::new();
s.replace(vec![
note("Notes/a.md"),
note("Notes/deep/b.md"),
note("Notesy/c.md"),
note("Other/d.md"),
]);
let dropped = s.apply_batch(vec![], &[], &["Notes".to_string()]);
assert_eq!(dropped, 2);
assert!(s.get("Notes/a.md").is_none());
assert!(s.get("Notes/deep/b.md").is_none());
// A sibling sharing the prefix as a string, but not as a directory.
assert!(s.get("Notesy/c.md").is_some());
assert!(s.get("Other/d.md").is_some());
}
#[test]
fn note_delete_and_prefix_delete_of_one_path_are_disjoint() {
// A vanished `*.md` path sends both, since it may have been a
// directory [ANW-40]. The note delete takes the key `Archive.md`,
// the prefix delete takes the keys under `Archive.md/`; neither
// reaches anything else.
let s = NoteStore::new();
s.replace(vec![
note("Archive.md"),
note("Archive.md/inner.md"),
note("Keep.md"),
]);
let dropped = s.apply_batch(
vec![],
&["Archive.md".to_string()],
&["Archive.md".to_string()],
);
assert_eq!(dropped, 1);
assert!(s.get("Archive.md").is_none());
assert!(s.get("Archive.md/inner.md").is_none());
assert!(s.get("Keep.md").is_some());
}
#[test]
fn apply_batch_prefix_delete_precedes_upserts() {
let s = NoteStore::new();
s.replace(vec![note("Notes/gone.md")]);
// A directory removed and recreated inside one window: only what the
// walk found survives.
s.apply_batch(vec![note("Notes/fresh.md")], &[], &["Notes".to_string()]);
assert!(s.get("Notes/gone.md").is_none());
assert!(s.get("Notes/fresh.md").is_some());
}
#[test]
fn upsert_replaces_by_path() {
let s = NoteStore::new();

881
src/telemetry.rs Normal file
View file

@ -0,0 +1,881 @@
//! Request-level telemetry for the HTTP surface, per
//! [ANW-37](https://crvrs.youtrack.cloud/issue/ANW-37).
//!
//! Two halves:
//!
//! - **Metrics** (every request): a request counter and a response-bytes
//! counter keyed by route, status, and conditional-GET outcome, plus a
//! duration histogram keyed by route and status. Exported over OTLP.
//! - **Traces** (slow or failing requests only): the incoming W3C
//! `traceparent` is extracted, and a request at or over the configured
//! threshold, or answering a 5xx, is recorded as a server span nested in
//! the propagated context. Every other request stays metrics-only, so the
//! ~3.6k requests/min steady state does not drown the trace backend.
//!
//! Transport configuration comes from the standard `OTEL_EXPORTER_OTLP_*`
//! 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
//! exactly as it did before this module existed. External installs run
//! unchanged.
use std::time::{Duration, SystemTime};
use anyhow::{Context as _, bail};
use axum::http::HeaderMap;
use opentelemetry::KeyValue;
use opentelemetry::metrics::{Counter, Histogram, MeterProvider as _};
use opentelemetry::propagation::{Extractor, TextMapPropagator};
use opentelemetry::trace::{Span, SpanKind, Tracer, TracerProvider as _};
use opentelemetry_otlp::{MetricExporter, SpanExporter};
use opentelemetry_sdk::Resource;
use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
use opentelemetry_sdk::propagation::TraceContextPropagator;
use opentelemetry_sdk::trace::Sampler;
use opentelemetry_sdk::trace::{SdkTracer, SdkTracerProvider};
use opentelemetry_semantic_conventions::resource::{SERVICE_NAME, SERVICE_VERSION};
/// The removed telemetry options, still parsed so their presence is an error
/// rather than a silent config drop on upgrade
/// ([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)]
pub struct RawTelemetryArgs {
/// Removed `--uptrace-dsn` / `ANWESEN_UPTRACE_DSN`.
pub uptrace_dsn: Option<String>,
/// Removed `--otlp-endpoint` / `ANWESEN_OTLP_ENDPOINT`.
pub otlp_endpoint: Option<String>,
/// Removed `--otlp-header` / `ANWESEN_OTLP_HEADERS`.
pub otlp_headers: Vec<String>,
/// `--otlp-slow-request-ms` / `ANWESEN_OTLP_SLOW_REQUEST_MS`. Kept: it
/// decides when anwesen emits a span, not where the export goes, and no
/// standard variable covers it.
pub slow_request_ms: u64,
}
/// The OTLP transport variables the SDK reads, captured once at startup so
/// the telemetry-on decision and the startup checks stay pure functions of
/// 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
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TelemetryConfig {
/// A request at or over this duration, or answering a 5xx, is recorded
/// as a server span.
pub slow_request: Duration,
/// The transport variables, kept for the startup log line only.
env: OtelEnv,
}
impl TelemetryConfig {
/// Resolve the surviving option plus the OTLP transport variables into an
/// optional config.
///
/// - A removed flag or `ANWESEN_` variable is an error naming its `OTEL_`
/// replacement: a deployment exporting today must fail one restart
/// rather than go quiet.
/// - No endpoint variable set means telemetry is off: `Ok(None)`.
/// - A query or fragment on `OTEL_EXPORTER_OTLP_ENDPOINT` is an error.
/// - A protocol this binary cannot speak is an error.
///
/// # Errors
/// Returns an error when a removed option is present, when the generic
/// endpoint carries a query or a fragment, or when a protocol variable
/// asks for anything but `http/protobuf`.
pub fn resolve(raw: &RawTelemetryArgs, env: OtelEnv) -> anyhow::Result<Option<Self>> {
check_removed(raw)?;
if !env.any_endpoint() {
return 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
/// replaces it. Silence is the expensive failure here: an upgrade that drops
/// the export config would stop telemetry with nothing in the log.
fn check_removed(raw: &RawTelemetryArgs) -> anyhow::Result<()> {
let removed: [(&str, &str, bool); 3] = [
(
"--uptrace-dsn / ANWESEN_UPTRACE_DSN",
"OTEL_EXPORTER_OTLP_ENDPOINT=https://api.uptrace.dev plus \
OTEL_EXPORTER_OTLP_HEADERS=uptrace-dsn=<the DSN, verbatim>",
raw.uptrace_dsn.is_some(),
),
(
"--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");
}
}
Ok(())
}
/// Reject a query or fragment on `OTEL_EXPORTER_OTLP_ENDPOINT`. The SDK
/// appends the per-signal path to this value textually, so a `?grpc=4317`
/// tail sends `POST /?grpc=4317/v1/metrics` and a `#frag` tail sends
/// `POST /` -- both dead exports, neither logged. Measured against a sink on
/// opentelemetry-otlp 0.32 ([ANW-42]). A path prefix composes correctly and
/// is left alone.
///
/// [ANW-42]: https://crvrs.youtrack.cloud/issue/ANW-42
fn check_generic_endpoint(endpoint: &str) -> anyhow::Result<()> {
if let Some(bad) = endpoint.find(['?', '#']) {
let tail = &endpoint[bad..];
bail!(
"OTEL_EXPORTER_OTLP_ENDPOINT {endpoint:?} has a trailing {tail:?}; \
the exporter would append the signal path after it and export nowhere. \
Pass the base URL alone, and put an uptrace DSN in \
OTEL_EXPORTER_OTLP_HEADERS=uptrace-dsn=<the DSN, verbatim>"
);
}
Ok(())
}
/// 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
/// template on purpose: `/notes/{*path}` serves both a note fetch and a
/// folder listing, and "304 share of note fetches" needs the two apart. The
/// bucket is a function of the request path prefix and its trailing slash.
#[must_use]
pub fn classify_route(path: &str) -> &'static str {
if path == "/health" {
"health"
} else if path == "/query" {
"query"
} else if path == "/notes/" {
"folder"
} else if path.starts_with("/notes/") {
if path.ends_with('/') {
"folder"
} else {
"note"
}
} else {
"other"
}
}
/// Whether a completed request additionally warrants a server span: it took
/// at least the slow threshold, or answered a server error. Everything else
/// stays metrics-only so the steady-state request rate does not flood the
/// trace backend.
#[must_use]
pub fn should_span(duration: Duration, slow_request: Duration, status: u16) -> bool {
duration >= slow_request || status >= 500
}
/// Conditional-GET outcome for the `conditional_get` label, derived from the
/// response status and whether the request carried `If-None-Match`:
///
/// - `not_modified`: a 304 (the client's etag matched);
/// - `revalidated`: `If-None-Match` was present but the body was still sent
/// (etag mismatch);
/// - `unconditional`: no `If-None-Match` header.
///
/// Together these answer both the 304 share of note fetches and the
/// If-None-Match presence share.
#[must_use]
pub fn conditional_get(status: u16, if_none_match_present: bool) -> &'static str {
if status == 304 {
"not_modified"
} else if if_none_match_present {
"revalidated"
} else {
"unconditional"
}
}
/// Histogram bucket boundaries for `http.server.request.duration`, in
/// seconds. Extended out to 600 s because the fleet p50 is ~305 s (ANW-38);
/// the default `OTel` buckets top out near 10 s, so every slow request would
/// land in one overflow bucket and the percentiles the issue needs would be
/// unreadable.
pub const DURATION_BUCKETS_SECONDS: &[f64] = &[
0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0, 600.0,
];
/// The W3C trace-context header values carried by a request, captured before
/// the handler consumes it. Extracting the parent context is deferred to
/// span-emission time, so the hot path (metrics-only requests) pays only two
/// header reads, not a full propagator extraction.
#[derive(Debug, Default, Clone)]
pub struct TraceHeaders {
pub traceparent: Option<String>,
pub tracestate: Option<String>,
}
impl TraceHeaders {
#[must_use]
pub fn from_headers(headers: &HeaderMap) -> Self {
let get = |name: &str| {
headers
.get(name)
.and_then(|v| v.to_str().ok())
.map(str::to_owned)
};
Self {
traceparent: get("traceparent"),
tracestate: get("tracestate"),
}
}
}
/// Lets the W3C propagator read the captured header values directly, without
/// rebuilding a `HeaderMap`. Only `traceparent` and `tracestate` matter for
/// trace-context extraction.
impl Extractor for TraceHeaders {
fn get(&self, key: &str) -> Option<&str> {
match key {
"traceparent" => self.traceparent.as_deref(),
"tracestate" => self.tracestate.as_deref(),
_ => None,
}
}
fn keys(&self) -> Vec<&str> {
let mut keys = Vec::with_capacity(2);
if self.traceparent.is_some() {
keys.push("traceparent");
}
if self.tracestate.is_some() {
keys.push("tracestate");
}
keys
}
}
/// Live telemetry handle: owns the OTLP meter and tracer providers and the
/// instruments, and records one observation per request. Created by [`init`]
/// only when telemetry is configured; when it is not, the request middleware
/// is never installed and this type is never constructed.
pub struct Telemetry {
inner: TelemetryInner,
}
impl Telemetry {
/// Record one completed request: bump the metric instruments, and -- when
/// the request was at or over the slow threshold or answered a 5xx --
/// emit a server span nested under the propagated trace context.
///
/// `start` is the wall-clock instant the request arrived, used as the
/// span start time so the span's own duration matches `duration`.
#[allow(clippy::too_many_arguments)]
pub fn finish(
&self,
route: &'static str,
method: &str,
status: u16,
if_none_match_present: bool,
body_bytes: u64,
duration: Duration,
start: SystemTime,
trace: &TraceHeaders,
) {
self.inner
.record_metrics(route, status, if_none_match_present, body_bytes, duration);
if should_span(duration, self.inner.slow_request, status) {
self.inner
.record_span(route, method, status, duration, start, trace);
}
}
/// Flush and shut down the providers. Called once at server exit.
pub fn shutdown(&self) {
self.inner.shutdown();
}
}
/// Build the telemetry handle from a resolved config, standing up the OTLP
/// meter and tracer providers.
///
/// # Errors
/// Returns an error when an OTLP exporter cannot be constructed.
pub fn init(config: TelemetryConfig) -> anyhow::Result<Telemetry> {
Ok(Telemetry {
inner: TelemetryInner::new(config)?,
})
}
// -- OTLP wiring seam -------------------------------------------------------
//
// `TelemetryInner` owns the OTel providers and instruments. The public
// surface above (`finish`, `shutdown`, `TraceHeaders`, `classify_route`,
// `conditional_get`, the bucket boundaries) is stable; only the bodies below
// change as the exporters are wired.
struct TelemetryInner {
slow_request: Duration,
meter_provider: SdkMeterProvider,
tracer_provider: SdkTracerProvider,
tracer: SdkTracer,
requests: Counter<u64>,
response_bytes: Counter<u64>,
duration: Histogram<f64>,
propagator: TraceContextPropagator,
}
impl TelemetryInner {
fn new(config: TelemetryConfig) -> anyhow::Result<Self> {
let TelemetryConfig { slow_request, env } = config;
// Service identity is a default only: a key OTEL_SERVICE_NAME or
// OTEL_RESOURCE_ATTRIBUTES already carries is left to the SDK's own
// detectors, because a builder attribute would win over them.
let mut resource = Resource::builder();
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.
// 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()
.with_http()
.build()
.context("build OTLP metric exporter")?;
let reader = PeriodicReader::builder(metric_exporter)
.with_interval(Duration::from_secs(15))
.build();
let meter_provider = SdkMeterProvider::builder()
.with_resource(resource.clone())
.with_reader(reader)
.build();
let meter = meter_provider.meter("anwesen");
let requests = meter
.u64_counter("http.server.requests")
.with_unit("{request}")
.with_description("HTTP requests by route, status, and conditional-GET outcome")
.build();
let response_bytes = meter
.u64_counter("http.server.response.body.size")
.with_unit("By")
.with_description(
"HTTP response body bytes by route, status, and conditional-GET outcome",
)
.build();
let duration = meter
.f64_histogram("http.server.request.duration")
.with_unit("s")
.with_description("HTTP request duration by route and status")
.with_boundaries(DURATION_BUCKETS_SECONDS.to_vec())
.build();
// -- traces: thread-based batch processor over an OTLP/HTTP exporter.
let span_exporter = SpanExporter::builder()
.with_http()
.build()
.context("build OTLP span exporter")?;
let tracer_provider = SdkTracerProvider::builder()
.with_resource(resource)
// Span creation is already gated to slow/5xx requests in
// `record_span`, so sample everything we choose to build.
.with_sampler(Sampler::AlwaysOn)
.with_batch_exporter(span_exporter)
.build();
let tracer = tracer_provider.tracer("anwesen");
tracing::info!(
endpoint = %env.describe(),
slow_request_ms = slow_request.as_millis(),
"telemetry: OTLP export enabled"
);
Ok(Self {
slow_request,
meter_provider,
tracer_provider,
tracer,
requests,
response_bytes,
duration,
propagator: TraceContextPropagator::new(),
})
}
fn record_metrics(
&self,
route: &'static str,
status: u16,
if_none_match_present: bool,
body_bytes: u64,
duration: Duration,
) {
let attrs = [
KeyValue::new("http.route", route),
KeyValue::new("http.response.status_code", i64::from(status)),
KeyValue::new(
"conditional_get",
conditional_get(status, if_none_match_present),
),
];
self.requests.add(1, &attrs);
self.response_bytes.add(body_bytes, &attrs);
// The histogram stays at route x status per the contract; the
// conditional-GET dimension lives on the counters only.
self.duration.record(
duration.as_secs_f64(),
&[
KeyValue::new("http.route", route),
KeyValue::new("http.response.status_code", i64::from(status)),
],
);
}
fn record_span(
&self,
route: &'static str,
method: &str,
status: u16,
duration: Duration,
start: SystemTime,
trace: &TraceHeaders,
) {
let parent_cx = self.propagator.extract(trace);
let mut span = self
.tracer
.span_builder(format!("{method} {route}"))
.with_kind(SpanKind::Server)
.with_start_time(start)
.with_attributes(vec![
KeyValue::new("http.request.method", method.to_string()),
KeyValue::new("http.route", route),
KeyValue::new("http.response.status_code", i64::from(status)),
])
.start_with_context(&self.tracer, &parent_cx);
span.end_with_timestamp(start + duration);
}
fn shutdown(&self) {
if let Err(e) = self.meter_provider.shutdown() {
tracing::warn!(error = %e, "telemetry: meter provider shutdown");
}
if let Err(e) = self.tracer_provider.shutdown() {
tracing::warn!(error = %e, "telemetry: tracer provider shutdown");
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn raw() -> RawTelemetryArgs {
RawTelemetryArgs::default()
}
fn generic(endpoint: &str) -> OtelEnv {
OtelEnv {
endpoint: Some(endpoint.into()),
..OtelEnv::default()
}
}
#[test]
fn resolve_off_when_no_endpoint_variable_is_set() {
let cfg = TelemetryConfig::resolve(&raw(), OtelEnv::default()).unwrap();
assert!(cfg.is_none(), "no OTEL_ endpoint means telemetry off");
}
#[test]
fn resolve_on_for_each_endpoint_variable() {
let per_signal = |field: fn(&mut OtelEnv)| {
let mut env = OtelEnv::default();
field(&mut env);
env
};
for env in [
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]);
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]
fn resolve_rejects_the_removed_uptrace_dsn() {
let err = TelemetryConfig::resolve(
&RawTelemetryArgs {
uptrace_dsn: Some("https://tok@api.uptrace.dev".into()),
..raw()
},
OtelEnv::default(),
)
.unwrap_err()
.to_string();
assert!(err.contains("--uptrace-dsn"), "{err}");
assert!(err.contains("OTEL_EXPORTER_OTLP_HEADERS"), "{err}");
}
#[test]
fn resolve_rejects_the_removed_otlp_endpoint() {
let err = TelemetryConfig::resolve(
&RawTelemetryArgs {
otlp_endpoint: Some("https://collector.example.com".into()),
..raw()
},
OtelEnv::default(),
)
.unwrap_err()
.to_string();
assert!(err.contains("--otlp-endpoint"), "{err}");
assert!(err.contains("OTEL_EXPORTER_OTLP_ENDPOINT"), "{err}");
}
#[test]
fn resolve_rejects_the_removed_otlp_header() {
let err = TelemetryConfig::resolve(
&RawTelemetryArgs {
otlp_headers: vec!["authorization=Bearer xyz".into()],
..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();
assert!(err.to_string().contains("removed"), "{err}");
}
#[test]
fn classify_route_buckets() {
assert_eq!(classify_route("/health"), "health");
assert_eq!(classify_route("/query"), "query");
assert_eq!(classify_route("/notes/"), "folder");
assert_eq!(classify_route("/notes/a.md"), "note");
assert_eq!(classify_route("/notes/Projects/a.md"), "note");
assert_eq!(classify_route("/notes/Projects/"), "folder");
assert_eq!(classify_route("/notes/Projects/anwesen/"), "folder");
assert_eq!(classify_route("/"), "other");
assert_eq!(classify_route("/favicon.ico"), "other");
}
#[test]
fn should_span_on_slow_or_5xx() {
let threshold = Duration::from_millis(500);
// Fast + success: metrics-only.
assert!(!should_span(Duration::from_millis(10), threshold, 200));
assert!(!should_span(Duration::from_millis(499), threshold, 304));
// At or over the threshold: span.
assert!(should_span(Duration::from_millis(500), threshold, 200));
assert!(should_span(Duration::from_secs(305), threshold, 200));
// 5xx spans even when fast.
assert!(should_span(Duration::from_millis(1), threshold, 500));
assert!(should_span(Duration::from_millis(1), threshold, 503));
// 4xx is a client error, not a server span trigger on its own.
assert!(!should_span(Duration::from_millis(1), threshold, 404));
}
#[test]
fn conditional_get_outcomes() {
assert_eq!(conditional_get(304, true), "not_modified");
// A 304 with no header cannot happen in practice, but the status
// wins: it is still "not modified".
assert_eq!(conditional_get(304, false), "not_modified");
assert_eq!(conditional_get(200, true), "revalidated");
assert_eq!(conditional_get(200, false), "unconditional");
assert_eq!(conditional_get(404, false), "unconditional");
}
}

View file

@ -154,12 +154,23 @@ pub enum ScanWarningKind {
/// Walk the vault and return every readable Markdown note alongside any
/// per-file issues. The walk never panics on a single broken file.
#[must_use]
pub fn scan(vault_root: &Path) -> ScanResult {
scan_from(vault_root, vault_root)
}
/// Walk `start` (a directory inside `vault_root`) and return every readable
/// Markdown note under it. Note paths stay relative to `vault_root`, so a
/// subtree result is directly comparable with a full [`scan`]. Used by the
/// watcher when a directory appears or is renamed into the vault [ANW-36]:
/// the native event names the directory, never its files.
#[must_use]
pub fn scan_from(vault_root: &Path, start: &Path) -> ScanResult {
let mut notes = Vec::new();
let mut issues = Vec::new();
let mut warnings = Vec::new();
let walker = WalkDir::new(vault_root)
let walker = WalkDir::new(start)
.follow_links(false)
.into_iter()
// The root entry itself may have a dot-prefixed name (e.g., a

View file

@ -4,7 +4,7 @@
//! [`notify::RecommendedWatcher`] over the vault root. Each native event is
//! pushed into a Tokio channel; [`run_debouncer`] drains the channel,
//! classifies events into [`WatchAction`]s, coalesces a 100 ms window's
//! worth into one [`Batch`], and casts the batch to the `index_writer`
//! worth into one [`WatchBatch`], and casts the batch to the `index_writer`
//! named process for a single [`crate::store::NoteStore`] write.
//!
//! See [[ADR-003 Filesystem Change Tracking]] for the event model.
@ -27,11 +27,17 @@ use crate::vault;
/// One path-scoped action derived from a native filesystem event. Always
/// 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
/// name only the directory when one is created, removed, or renamed; the
/// files under it produce no events of their own [ANW-36].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WatchAction {
Upsert(String),
Delete(String),
UpsertTree(String),
DeleteTree(String),
}
/// One debouncer window's worth of coalesced changes.
@ -39,6 +45,10 @@ pub enum WatchAction {
pub struct WatchBatch {
pub upserts: Vec<String>,
pub deletes: Vec<String>,
/// Directories to walk for notes to upsert.
pub upsert_trees: Vec<String>,
/// Directories whose indexed notes are all gone.
pub delete_trees: Vec<String>,
}
/// Classify one [`Event`] into zero or more [`WatchAction`]s. Dot-segments
@ -54,60 +64,113 @@ pub fn map_event(event: &Event, vault_root: &Path) -> Vec<WatchAction> {
let mut actions = Vec::new();
let action_kind = classify(event.kind);
match action_kind {
Some(EventAction::Upsert) => {
Some(EventAction::Modify) => {
// Content and metadata events only ever name a file.
for p in &event.paths {
if let Some(rel) = vault_relative(vault_root, p) {
if let Some(rel) = note_relative(vault_root, p) {
actions.push(WatchAction::Upsert(rel));
}
}
}
Some(EventAction::Delete) => {
Some(EventAction::Appear) => {
for p in &event.paths {
if let Some(rel) = vault_relative(vault_root, p) {
actions.push(WatchAction::Delete(rel));
}
actions.extend(appear(vault_root, p));
}
}
Some(EventAction::Vanish) => {
for p in &event.paths {
actions.extend(vanish(vault_root, p));
}
}
Some(EventAction::Rename) => {
// Notify packs (from, to) in event.paths in that order.
if let Some(from) = event.paths.first()
&& let Some(rel) = vault_relative(vault_root, from)
{
actions.push(WatchAction::Delete(rel));
if let Some(from) = event.paths.first() {
actions.extend(vanish(vault_root, from));
}
if let Some(to) = event.paths.get(1)
&& let Some(rel) = vault_relative(vault_root, to)
{
actions.push(WatchAction::Upsert(rel));
if let Some(to) = event.paths.get(1) {
actions.extend(appear(vault_root, to));
}
}
Some(EventAction::RenameUnpaired) => {
// FSEvents (macOS) cannot pair the two sides of a rename and
// reports `Modify(Name(Any))` for each side separately, so the
// direction has to come off the filesystem [ANW-36].
for p in &event.paths {
if p.exists() {
actions.extend(appear(vault_root, p));
} else {
actions.extend(vanish(vault_root, p));
}
}
// `p.exists()` can only report the moment it is asked. A path
// that vanishes right after the probe is caught downstream:
// `build_index_batch` turns a not-found upsert into a delete.
}
None => {}
}
actions
}
/// A path that now exists: a note to read, or a directory to walk. The
/// filesystem answers which, so a non-note file (an attachment, an editor
/// temp file) costs nothing beyond the probe.
fn appear(vault_root: &Path, abs: &Path) -> Option<WatchAction> {
if abs.is_dir() {
tree_relative(vault_root, abs).map(WatchAction::UpsertTree)
} else {
note_relative(vault_root, abs).map(WatchAction::Upsert)
}
}
/// A path that is gone: a note to drop, or a directory whose notes are all
/// gone with it. The filesystem cannot be asked -- it no longer holds the
/// entry -- so both are emitted and the index decides which one matches.
///
/// A `.md` suffix does not prove the path was a file: a directory may carry
/// it too, and then the suffix test alone strands its notes in the index
/// [ANW-40]. The two key sets are disjoint -- the note delete removes the key
/// `dir`, the prefix delete removes keys under `dir/` -- so emitting both is
/// always safe. It costs one range scan per deleted note.
fn vanish(vault_root: &Path, abs: &Path) -> Vec<WatchAction> {
let mut actions = Vec::new();
if let Some(rel) = note_relative(vault_root, abs) {
actions.push(WatchAction::Delete(rel));
}
if let Some(rel) = tree_relative(vault_root, abs) {
actions.push(WatchAction::DeleteTree(rel));
}
actions
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum EventAction {
Upsert,
Delete,
/// Existing file, new content.
Modify,
/// Path came into the vault.
Appear,
/// Path left the vault.
Vanish,
/// Both sides in one event, `(from, to)`.
Rename,
/// One side of a rename, direction unknown.
RenameUnpaired,
}
fn classify(kind: EventKind) -> Option<EventAction> {
match kind {
EventKind::Create(_)
| EventKind::Modify(
ModifyKind::Data(_)
| ModifyKind::Metadata(_)
| ModifyKind::Any
| ModifyKind::Name(RenameMode::To),
)
| EventKind::Access(AccessKind::Close(AccessMode::Write)) => Some(EventAction::Upsert),
EventKind::Modify(ModifyKind::Data(_) | ModifyKind::Metadata(_) | ModifyKind::Any)
| EventKind::Access(AccessKind::Close(AccessMode::Write)) => Some(EventAction::Modify),
EventKind::Create(_) | EventKind::Modify(ModifyKind::Name(RenameMode::To)) => {
Some(EventAction::Appear)
}
EventKind::Modify(ModifyKind::Name(RenameMode::From)) | EventKind::Remove(_) => {
Some(EventAction::Delete)
Some(EventAction::Vanish)
}
EventKind::Modify(ModifyKind::Name(RenameMode::Both)) => Some(EventAction::Rename),
// Access(non-close), Modify(Name(Any|Other)), Other, Any -> drop.
EventKind::Modify(ModifyKind::Name(RenameMode::Any | RenameMode::Other)) => {
Some(EventAction::RenameUnpaired)
}
// Access(non-close), Other, Any -> drop.
_ => None,
}
}
@ -120,8 +183,10 @@ pub fn is_overflow(event: &Event) -> bool {
}
/// Collapse a sequence of [`WatchAction`]s into a single batch. The last
/// action per path wins (delete-then-upsert ends up as upsert, and so on).
/// The batch keeps deterministic order by sorting paths inside each list.
/// action per path wins (delete-then-upsert ends up as upsert, and so on);
/// notes and directories are tracked separately, since a directory action
/// covers paths a note action cannot name. The batch keeps deterministic
/// order by sorting paths inside each list.
#[must_use]
pub fn coalesce(actions: impl IntoIterator<Item = WatchAction>) -> WatchBatch {
#[derive(Clone, Copy)]
@ -129,36 +194,64 @@ pub fn coalesce(actions: impl IntoIterator<Item = WatchAction>) -> WatchBatch {
Upsert,
Delete,
}
let mut state: BTreeMap<String, Last> = BTreeMap::new();
let mut notes: BTreeMap<String, Last> = BTreeMap::new();
let mut trees: BTreeMap<String, Last> = BTreeMap::new();
for a in actions {
match a {
WatchAction::Upsert(p) => {
state.insert(p, Last::Upsert);
notes.insert(p, Last::Upsert);
}
WatchAction::Delete(p) => {
state.insert(p, Last::Delete);
notes.insert(p, Last::Delete);
}
WatchAction::UpsertTree(p) => {
trees.insert(p, Last::Upsert);
}
WatchAction::DeleteTree(p) => {
trees.insert(p, Last::Delete);
}
}
}
let mut upserts = Vec::new();
let mut deletes = Vec::new();
for (path, last) in state {
let mut batch = WatchBatch::default();
for (path, last) in notes {
match last {
Last::Upsert => upserts.push(path),
Last::Delete => deletes.push(path),
Last::Upsert => batch.upserts.push(path),
Last::Delete => batch.deletes.push(path),
}
}
WatchBatch { upserts, deletes }
for (path, last) in trees {
match last {
Last::Upsert => batch.upsert_trees.push(path),
Last::Delete => batch.delete_trees.push(path),
}
}
batch
}
fn is_markdown(path: &Path) -> bool {
path.extension().is_some_and(|e| e == "md")
}
/// Filter and normalize an absolute notify path. Returns the vault-relative
/// forward-slash path if the entry is a `.md` file outside any dot-directory;
/// returns `None` otherwise (so the caller drops the event).
fn vault_relative(vault_root: &Path, abs: &Path) -> Option<String> {
let rel = abs.strip_prefix(vault_root).ok()?;
if rel.extension().is_none_or(|e| e != "md") {
fn note_relative(vault_root: &Path, abs: &Path) -> Option<String> {
if !is_markdown(abs) {
return None;
}
relative(vault_root, abs)
}
/// Same, for a directory: any path that is not a note. The vault root itself
/// relativizes to the empty string and is dropped -- a root-level event is a
/// vault-wide signal the rescan path handles, not a prefix to delete.
fn tree_relative(vault_root: &Path, abs: &Path) -> Option<String> {
let rel = relative(vault_root, abs)?;
if rel.is_empty() { None } else { Some(rel) }
}
fn relative(vault_root: &Path, abs: &Path) -> Option<String> {
let rel = abs.strip_prefix(vault_root).ok()?;
for component in rel.components() {
let s = component.as_os_str().to_str()?;
if s.starts_with('.') {
@ -226,7 +319,11 @@ pub async fn run_debouncer(
);
}
let batch = coalesce(actions);
if batch.upserts.is_empty() && batch.deletes.is_empty() {
if batch.upserts.is_empty()
&& batch.deletes.is_empty()
&& batch.upsert_trees.is_empty()
&& batch.delete_trees.is_empty()
{
continue;
}
let index_batch = build_index_batch(&vault_root, batch);
@ -239,10 +336,17 @@ pub async fn run_debouncer(
fn build_index_batch(vault_root: &Path, batch: WatchBatch) -> IndexBatch {
let mut upserts = Vec::with_capacity(batch.upserts.len());
let mut deletes = batch.deletes;
for rel in batch.upserts {
let abs = absolute_path(vault_root, &rel);
match vault::scan_one(vault_root, &abs) {
Ok(note) => upserts.push(note),
// The file is gone by the time we read it. An event backend that
// coalesces create-and-remove into one upsert-shaped event would
// otherwise leave the entry in the index forever [ANW-36].
Err(vault::ScanIssueKind::Io(e)) if e.kind() == std::io::ErrorKind::NotFound => {
deletes.push(rel);
}
Err(kind) => {
tracing::warn!(
path = %abs.display(),
@ -252,9 +356,29 @@ fn build_index_batch(vault_root: &Path, batch: WatchBatch) -> IndexBatch {
}
}
}
// A walked directory carries its own prefix delete: the index under that
// prefix must match what the walk found. Without it, a directory renamed
// out and another renamed in within one window coalesces to a bare
// `UpsertTree` -- the delete is lost and the old notes outlive their
// files [ANW-36].
let mut delete_prefixes = batch.delete_trees;
delete_prefixes.extend(batch.upsert_trees.iter().cloned());
for dir in &batch.upsert_trees {
let abs = absolute_path(vault_root, dir);
let result = vault::scan_from(vault_root, &abs);
for issue in &result.issues {
tracing::warn!(
path = %issue.path.display(),
error = %issue.kind,
"filesystem_watcher: subtree walk issue; skipping note"
);
}
upserts.extend(result.notes);
}
IndexBatch {
upserts,
deletes: batch.deletes,
deletes,
delete_prefixes,
}
}
@ -339,7 +463,10 @@ mod tests {
);
assert_eq!(
map_event(&e, &vault()),
vec![WatchAction::Delete("a.md".into())]
vec![
WatchAction::Delete("a.md".into()),
WatchAction::DeleteTree("a.md".into())
]
);
}
@ -365,6 +492,7 @@ mod tests {
map_event(&e, &vault()),
vec![
WatchAction::Delete("a.md".into()),
WatchAction::DeleteTree("a.md".into()),
WatchAction::Upsert("b.md".into())
]
);
@ -375,7 +503,25 @@ mod tests {
let e = ev(EventKind::Remove(RemoveKind::File), &["/v/a.md"]);
assert_eq!(
map_event(&e, &vault()),
vec![WatchAction::Delete("a.md".into())]
vec![
WatchAction::Delete("a.md".into()),
WatchAction::DeleteTree("a.md".into())
]
);
}
#[test]
fn removed_md_directory_drops_the_notes_under_it() {
// A directory may be named `*.md` too. The vanished path cannot be
// probed, so the note delete and the prefix delete both go out and
// the index applies whichever matches [ANW-40].
let e = ev(EventKind::Remove(RemoveKind::Folder), &["/v/Archive.md"]);
assert_eq!(
map_event(&e, &vault()),
vec![
WatchAction::Delete("Archive.md".into()),
WatchAction::DeleteTree("Archive.md".into())
]
);
}
@ -393,6 +539,145 @@ mod tests {
assert_eq!(batch.deletes, vec!["c.md".to_string()]);
}
#[test]
fn coalesce_keeps_notes_and_trees_apart() {
let actions = vec![
WatchAction::Upsert("Notes/a.md".into()),
WatchAction::DeleteTree("Notes".into()),
WatchAction::UpsertTree("Fresh".into()),
];
let batch = coalesce(actions);
assert_eq!(batch.upserts, vec!["Notes/a.md".to_string()]);
assert!(batch.deletes.is_empty());
assert_eq!(batch.upsert_trees, vec!["Fresh".to_string()]);
assert_eq!(batch.delete_trees, vec!["Notes".to_string()]);
}
#[test]
fn removed_directory_is_a_tree_delete() {
let e = ev(EventKind::Remove(RemoveKind::Folder), &["/v/Notes"]);
assert_eq!(
map_event(&e, &vault()),
vec![WatchAction::DeleteTree("Notes".into())]
);
}
#[test]
fn directory_renamed_away_is_a_tree_delete() {
let e = ev(
EventKind::Modify(ModifyKind::Name(RenameMode::From)),
&["/v/Notes"],
);
assert_eq!(
map_event(&e, &vault()),
vec![WatchAction::DeleteTree("Notes".into())]
);
}
#[test]
fn vault_root_itself_is_not_a_tree_action() {
let e = ev(EventKind::Remove(RemoveKind::Folder), &["/v"]);
assert!(map_event(&e, &vault()).is_empty());
}
#[test]
fn removed_non_note_file_touches_no_note() {
// A vanished path with no `.md` suffix is treated as a directory;
// the prefix simply matches nothing in the store.
let e = ev(EventKind::Remove(RemoveKind::File), &["/v/image.png"]);
assert_eq!(
map_event(&e, &vault()),
vec![WatchAction::DeleteTree("image.png".into())]
);
}
#[test]
fn created_directory_is_a_tree_upsert() {
let root = tempfile::tempdir().unwrap();
std::fs::create_dir(root.path().join("Notes")).unwrap();
let e = ev(EventKind::Create(CreateKind::Folder), &[]).add_path(root.path().join("Notes"));
assert_eq!(
map_event(&e, root.path()),
vec![WatchAction::UpsertTree("Notes".into())]
);
}
#[test]
fn unpaired_rename_resolves_by_existence() {
// FSEvents (macOS) reports each side of a rename as
// `Modify(Name(Any))` with no way to pair them.
let root = tempfile::tempdir().unwrap();
std::fs::write(root.path().join("here.md"), "x").unwrap();
let kind = EventKind::Modify(ModifyKind::Name(RenameMode::Any));
let present = ev(kind, &[]).add_path(root.path().join("here.md"));
assert_eq!(
map_event(&present, root.path()),
vec![WatchAction::Upsert("here.md".into())]
);
let absent = ev(kind, &[]).add_path(root.path().join("gone.md"));
assert_eq!(
map_event(&absent, root.path()),
vec![
WatchAction::Delete("gone.md".into()),
WatchAction::DeleteTree("gone.md".into())
]
);
}
#[test]
fn upsert_of_a_vanished_file_becomes_a_delete() {
let root = tempfile::tempdir().unwrap();
let batch = WatchBatch {
upserts: vec!["gone.md".to_string()],
..WatchBatch::default()
};
let index_batch = build_index_batch(root.path(), batch);
assert!(index_batch.upserts.is_empty());
assert_eq!(index_batch.deletes, vec!["gone.md".to_string()]);
}
#[test]
fn tree_upsert_walks_the_subtree() {
let root = tempfile::tempdir().unwrap();
std::fs::create_dir_all(root.path().join("Notes/deep")).unwrap();
std::fs::write(root.path().join("Notes/a.md"), "---\nk: 1\n---\nbody\n").unwrap();
std::fs::write(root.path().join("Notes/deep/b.md"), "body\n").unwrap();
std::fs::write(root.path().join("Notes/skip.txt"), "no\n").unwrap();
std::fs::write(root.path().join("outside.md"), "no\n").unwrap();
let batch = WatchBatch {
upsert_trees: vec!["Notes".to_string()],
..WatchBatch::default()
};
let index_batch = build_index_batch(root.path(), batch);
let mut paths: Vec<String> = index_batch.upserts.into_iter().map(|n| n.path).collect();
paths.sort();
assert_eq!(paths, vec!["Notes/a.md", "Notes/deep/b.md"]);
}
#[test]
fn tree_upsert_carries_its_prefix_delete() {
// One directory renamed out and another renamed in within one window
// coalesces to a bare `UpsertTree`; the walk must still drop whatever
// the old directory left in the index [ANW-36].
let root = tempfile::tempdir().unwrap();
std::fs::create_dir_all(root.path().join("Notes")).unwrap();
std::fs::write(root.path().join("Notes/new.md"), "body\n").unwrap();
let batch = coalesce([
WatchAction::DeleteTree("Notes".to_string()),
WatchAction::UpsertTree("Notes".to_string()),
]);
assert!(batch.delete_trees.is_empty());
let index_batch = build_index_batch(root.path(), batch);
assert_eq!(index_batch.delete_prefixes, vec!["Notes".to_string()]);
let paths: Vec<String> = index_batch.upserts.into_iter().map(|n| n.path).collect();
assert_eq!(paths, vec!["Notes/new.md"]);
}
#[test]
fn overflow_event_returns_empty_actions() {
// Caller dispatches rescan_now; map_event itself produces no per-path actions.

View file

@ -54,6 +54,20 @@ if [[ $ready -ne 1 ]]; then
exit 1
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.
shopt -s globstar nullglob
files=("$ROOT"/tests/hurl/**/*.hurl)

View file

@ -0,0 +1,33 @@
//! `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"
);
}