chore(build): integrate PostgreSQL tracing foundation

Signed-off-by: Cea Stapleton Cordasco <261786559+cea@users.noreply.github.com>
This commit is contained in:
Cea Stapleton Cordasco
2026-08-11 17:49:29 -05:00
16 changed files with 559 additions and 117 deletions
-33
View File
@@ -69,39 +69,6 @@ RELAY_URL=ws://localhost:3000
# BUZZ_RATE_LIMIT_AGENT_ELEVATED_MESSAGES_PER_MIN=300
# BUZZ_RATE_LIMIT_AGENT_PLATFORM_MESSAGES_PER_MIN=600
# Relay-verified identity (disabled by default). When enabled, the relay
# requires authenticated requests to present a valid corporate JWT, then binds
# the configured uid claim to the Nostr pubkey proven by NIP-42/NIP-98. The JWT
# may be injected by a trusted proxy or attached by a first-party client; the
# relay treats both as the same header. Clients must forward the configured
# token header on every authenticated HTTP request and session handshake.
#
# Operational notes for the initial implementation:
# - When a trusted proxy injects this header, it MUST overwrite any inbound
# client-supplied value before forwarding to the relay.
# - Revocation and rotation are explicit database lifecycle operations;
# ordinary authentication never silently replaces a key.
# - JWKS outages fail closed for human JWT authentication. Delegated agent
# admission can still work when the owner binding is already present.
# - DISPLAY_CLAIM is private binding metadata. It is never projected publicly
# unless PUBLIC_DISPLAY_CLAIM is separately configured.
# BUZZ_REQUIRE_CORPORATE_IDENTITY=false
# BUZZ_CORPORATE_IDENTITY_JWT_HEADER=x-forwarded-identity-token
# BUZZ_ALLOW_CORPORATE_IDENTITY_DELEGATION=true
# When a request carries both a JWT and a verified NIP-OA owner declaration,
# choose whether the JWT identifies the signer or the owner binding delegates
# access. Defaults to direct; deployments that inject an owner's JWT into agent
# requests can explicitly select delegated.
# BUZZ_CORPORATE_IDENTITY_AUTH_PRECEDENCE=direct
# BUZZ_CORPORATE_IDENTITY_JWKS_URI=https://idp.example/.well-known/jwks.json
# BUZZ_CORPORATE_IDENTITY_ISSUER=https://idp.example
# BUZZ_CORPORATE_IDENTITY_AUDIENCE=buzz-relay
# BUZZ_CORPORATE_IDENTITY_UID_CLAIM=sub
# BUZZ_CORPORATE_IDENTITY_DISPLAY_CLAIM=email
# Optional, public NIP-85 label. Unset by default to keep identity claims private.
# BUZZ_CORPORATE_IDENTITY_PUBLIC_DISPLAY_CLAIM=display_name
# BUZZ_CORPORATE_IDENTITY_NPUB_CLAIM=buzz_npub
# -----------------------------------------------------------------------------
# Git (NIP-34 bare repositories)
# -----------------------------------------------------------------------------
Generated
+20
View File
@@ -916,6 +916,7 @@ name = "buzz-audit"
version = "0.1.0"
dependencies = [
"buzz-core",
"buzz-datastore-tracing",
"chrono",
"futures-util",
"hex",
@@ -1032,11 +1033,27 @@ dependencies = [
"zeroize",
]
[[package]]
name = "buzz-datastore-tracing"
version = "0.1.0"
dependencies = [
"opentelemetry 0.32.0",
"opentelemetry_sdk 0.32.1",
"proc-macro2",
"quote",
"syn 2.0.117",
"tokio",
"tracing",
"tracing-opentelemetry",
"tracing-subscriber",
]
[[package]]
name = "buzz-db"
version = "0.1.0"
dependencies = [
"buzz-core",
"buzz-datastore-tracing",
"chrono",
"hex",
"metrics",
@@ -1225,6 +1242,7 @@ dependencies = [
"buzz-auth",
"buzz-conformance",
"buzz-core",
"buzz-datastore-tracing",
"buzz-db",
"buzz-media",
"buzz-pubsub",
@@ -1322,9 +1340,11 @@ name = "buzz-search"
version = "0.1.0"
dependencies = [
"buzz-core",
"buzz-datastore-tracing",
"sqlx",
"thiserror 2.0.18",
"tokio",
"tracing",
"uuid",
]
+2
View File
@@ -28,6 +28,7 @@ members = [
"crates/buzz-dev-mcp",
"crates/buzz-voice",
"crates/buzz-backend-kubernetes",
"crates/buzz-datastore-tracing",
"examples/countdown-bot",
]
exclude = ["desktop/src-tauri"]
@@ -145,6 +146,7 @@ buzz-media = { path = "crates/buzz-media" }
buzz-sdk = { path = "crates/buzz-sdk" }
buzz-ws-client = { path = "crates/buzz-ws-client" }
buzz-relay-mesh = { path = "crates/buzz-relay-mesh" }
buzz-datastore-tracing = { path = "crates/buzz-datastore-tracing" }
# CI profile — builds the relay for desktop e2e. Dependencies keep full
# release optimization (warm from main's cache; they carry the runtime hot
+1
View File
@@ -9,6 +9,7 @@ description = "Hash-chain audit log for Buzz"
[dependencies]
buzz-core = { workspace = true }
buzz-datastore-tracing = { workspace = true }
sqlx = { workspace = true }
tokio = { workspace = true }
serde = { workspace = true }
+18 -5
View File
@@ -1,10 +1,11 @@
use chrono::{DateTime, Utc};
use futures_util::FutureExt as _;
use sqlx::{Acquire, PgPool, Row};
use tracing::{debug, instrument, warn};
use tracing::{debug, warn};
use uuid::Uuid;
use buzz_core::CommunityId;
use buzz_datastore_tracing::datastore_span;
use crate::{
action::AuditAction,
@@ -49,7 +50,11 @@ impl AuditService {
/// Serialized per-community via `pg_advisory_lock`. Postgres advisory locks
/// are session-scoped, so we acquire before the transaction and release
/// after commit (or on any error path).
#[instrument(skip(self, entry), fields(action = %entry.action))]
#[datastore_span(
name = "audit_log",
system = "postgresql",
fields(action = %entry.action)
)]
pub async fn log(&self, entry: NewAuditEntry) -> Result<AuditEntry, AuditError> {
let mut conn = self.pool.acquire().await?;
@@ -156,7 +161,11 @@ impl AuditService {
/// Reads exactly that community's chain — it can never observe another
/// community's entries or head. Returns `Ok(false)` if the range is empty,
/// `Ok(true)` if the segment is internally consistent.
#[instrument(skip(self))]
#[datastore_span(
name = "audit_verify_chain",
system = "postgresql",
fields(from_seq = from_seq, to_seq = to_seq)
)]
pub async fn verify_chain(
&self,
community: CommunityId,
@@ -208,7 +217,11 @@ impl AuditService {
/// Returns up to `limit` entries from one community's chain starting at
/// `from_seq`, ordered by sequence number. Scoped to `community` — never
/// returns another community's rows.
#[instrument(skip(self))]
#[datastore_span(
name = "audit_get_entries",
system = "postgresql",
fields(from_seq = from_seq, limit = limit)
)]
pub async fn get_entries(
&self,
community: CommunityId,
@@ -274,7 +287,7 @@ mod tests {
async fn test_pool() -> Option<PgPool> {
let url = std::env::var("DATABASE_URL")
.unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".into());
.unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".into()); // sadscan:disable np.postgres.1 -- local test fixture
PgPool::connect(&url).await.ok()
}
+24
View File
@@ -0,0 +1,24 @@
[package]
name = "buzz-datastore-tracing"
version.workspace = true
edition.workspace = true
rust-version.workspace = true
license.workspace = true
repository.workspace = true
description = "Privacy-preserving datastore tracing policy macros for Buzz"
[lib]
proc-macro = true
[dependencies]
proc-macro2 = "1"
quote = "1"
syn = { version = "2", features = ["full"] }
[dev-dependencies]
opentelemetry = { workspace = true }
opentelemetry_sdk = { workspace = true, features = ["testing"] }
tokio = { workspace = true }
tracing = { workspace = true }
tracing-opentelemetry = { workspace = true }
tracing-subscriber = { workspace = true }
+168
View File
@@ -0,0 +1,168 @@
//! Policy-enforcing instrumentation for logical datastore operations.
use proc_macro::TokenStream;
use proc_macro2::TokenStream as TokenStream2;
use quote::{format_ident, quote};
use syn::parse::{Parse, ParseStream};
use syn::{
parenthesized, parse_macro_input, Error, Ident, ItemFn, LitStr, Result, ReturnType, Token, Type,
};
struct DatastoreArgs {
name: LitStr,
system: LitStr,
fields: Option<TokenStream2>,
}
impl Parse for DatastoreArgs {
fn parse(input: ParseStream<'_>) -> Result<Self> {
let mut name = None;
let mut system = None;
let mut fields = None;
while !input.is_empty() {
let key: Ident = input.parse()?;
match key.to_string().as_str() {
"name" => {
if name.is_some() {
return Err(Error::new(key.span(), "duplicate `name` argument"));
}
input.parse::<Token![=]>()?;
name = Some(input.parse()?);
}
"system" => {
if system.is_some() {
return Err(Error::new(key.span(), "duplicate `system` argument"));
}
input.parse::<Token![=]>()?;
system = Some(input.parse()?);
}
"fields" => {
if fields.is_some() {
return Err(Error::new(key.span(), "duplicate `fields` argument"));
}
let content;
parenthesized!(content in input);
fields = Some(content.parse()?);
}
_ => {
return Err(Error::new(
key.span(),
"expected `name`, `system`, or `fields`",
))
}
}
if !input.is_empty() {
input.parse::<Token![,]>()?;
}
}
Ok(Self {
name: name.ok_or_else(|| Error::new(input.span(), "missing `name`"))?,
system: system.ok_or_else(|| Error::new(input.span(), "missing `system`"))?,
fields,
})
}
}
/// Instruments an async logical datastore operation according to Buzz policy.
///
/// PostgreSQL spans always omit function arguments, use the `buzz_datastore`
/// target, and expose only canonical semantic fields plus explicitly supplied
/// safe fields. An `Err` sets `otel.status_code` without inspecting the error.
#[proc_macro_attribute]
pub fn datastore_span(args: TokenStream, item: TokenStream) -> TokenStream {
let args = parse_macro_input!(args as DatastoreArgs);
let mut function = parse_macro_input!(item as ItemFn);
if function.sig.asyncness.is_none() {
return Error::new_spanned(
function.sig.fn_token,
"`datastore_span` requires an async function",
)
.into_compile_error()
.into();
}
if args.system.value() != "postgresql" {
return Error::new_spanned(
args.system,
"unsupported datastore system; only `postgresql` is currently supported",
)
.into_compile_error()
.into();
}
let name = args.name;
let extra_fields = args.fields.map(|tokens| quote!(, #tokens));
function.attrs.push(syn::parse_quote!(
#[::tracing::instrument(
target = "buzz_datastore",
name = #name,
skip_all,
fields(
otel.kind = "client",
db.system.name = "postgresql",
otel.status_code = ::tracing::field::Empty
#extra_fields
)
)]
));
let return_type: Type = match &function.sig.output {
ReturnType::Type(_, ty) => *ty.clone(),
ReturnType::Default => syn::parse_quote!(()),
};
let returns_result = match &return_type {
Type::Path(path) => path
.path
.segments
.last()
.is_some_and(|segment| segment.ident == "Result"),
_ => false,
};
let original_body = function.block;
let result = format_ident!("__buzz_datastore_result_7f3a9c");
let record_error = returns_result.then(|| {
quote! {
if #result.is_err() {
::tracing::Span::current().record("otel.status_code", "ERROR");
}
}
});
function.block = Box::new(syn::parse_quote!({
let #result: #return_type = (async #original_body).await;
#record_error
#result
}));
quote!(#function).into()
}
#[cfg(test)]
mod tests {
use super::DatastoreArgs;
#[test]
fn parses_safe_tracing_fields() {
let args: DatastoreArgs = syn::parse_str(
r#"name = "audit", system = "postgresql", fields(action = %entry.action, from_seq = from_seq, limit = limit)"#,
)
.expect("valid arguments");
assert!(args.fields.is_some());
}
#[test]
fn rejects_duplicate_arguments() {
for args in [
r#"name = "one", name = "two", system = "postgresql""#,
r#"name = "one", system = "postgresql", system = "postgresql""#,
r#"name = "one", system = "postgresql", fields(a = 1), fields(b = 2)"#,
] {
let error = match syn::parse_str::<DatastoreArgs>(args) {
Ok(_) => panic!("duplicate accepted"),
Err(error) => error,
};
assert!(error.to_string().starts_with("duplicate `"));
}
}
}
@@ -0,0 +1,80 @@
use buzz_datastore_tracing::datastore_span;
use opentelemetry::trace::{SpanKind, Status, TracerProvider as _};
use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider};
use tracing_subscriber::prelude::*;
const DIRECT_ERROR: &str = "raw-secret-direct-error";
const QUESTION_ERROR: &str = "raw-secret-question-error";
fn question_path(fail: bool) -> Result<(), &'static str> {
if fail {
Err(QUESTION_ERROR)
} else {
Ok(())
}
}
#[datastore_span(name = "test_operation", system = "postgresql", fields(limit = limit))]
async fn operation(
limit: usize,
direct_error: bool,
question_error: bool,
) -> Result<usize, &'static str> {
if direct_error {
return Err(DIRECT_ERROR);
}
question_path(question_error)?;
Ok(limit)
}
#[tokio::test(flavor = "current_thread")]
async fn exports_policy_fields_without_error_or_argument_data() {
let exporter = InMemorySpanExporter::default();
let provider = SdkTracerProvider::builder()
.with_simple_exporter(exporter.clone())
.build();
let subscriber = tracing_subscriber::registry()
.with(tracing_opentelemetry::layer().with_tracer(provider.tracer("datastore-macro-test")));
let _subscriber_guard = tracing::subscriber::set_default(subscriber);
assert_eq!(operation(7, false, false).await, Ok(7));
assert_eq!(operation(8, true, false).await, Err(DIRECT_ERROR));
assert_eq!(operation(9, false, true).await, Err(QUESTION_ERROR));
provider.force_flush().expect("spans flush");
let spans = exporter.get_finished_spans().expect("exported spans");
assert_eq!(spans.len(), 3);
for (span, (expected_limit, expected_status)) in spans.iter().zip([
(7_i64, Status::Unset),
(8_i64, Status::error("")),
(9_i64, Status::error("")),
]) {
assert_eq!(span.name, "test_operation");
assert_eq!(span.span_kind, SpanKind::Client);
assert_eq!(span.status, expected_status);
let attributes = span
.attributes
.iter()
.map(|attribute| (attribute.key.as_str(), attribute.value.to_string()))
.collect::<Vec<_>>();
assert!(attributes.contains(&("target", "buzz_datastore".to_owned())));
assert!(attributes.contains(&("db.system.name", "postgresql".to_owned())));
assert!(attributes.contains(&("limit", expected_limit.to_string())));
assert!(!attributes
.iter()
.any(|(key, _)| { matches!(*key, "direct_error" | "question_error") }));
let exported = format!("{span:?}");
assert!(!exported.contains(DIRECT_ERROR));
assert!(!exported.contains(QUESTION_ERROR));
assert!(span.events.iter().all(|event| {
!format!("{event:?}").contains(DIRECT_ERROR)
&& !format!("{event:?}").contains(QUESTION_ERROR)
}));
if let Status::Error { description } = &span.status {
assert!(description.is_empty());
}
}
}
+1
View File
@@ -9,6 +9,7 @@ description = "Postgres event store and data access layer for Buzz"
[dependencies]
buzz-core = { workspace = true }
buzz-datastore-tracing = { workspace = true }
sqlx = { workspace = true }
tokio = { workspace = true }
serde = { workspace = true }
File diff suppressed because it is too large Load Diff
+8
View File
@@ -63,6 +63,8 @@ use chrono::{DateTime, Utc};
use sqlx::{PgConnection, PgPool, Row};
use uuid::Uuid;
use buzz_datastore_tracing::datastore_span;
/// Seconds of `created_at` history the commit-time floor guard tolerates.
///
/// Must exceed the relay's ingest envelope (±900 s) by enough slack that a
@@ -315,6 +317,7 @@ impl ReplicaFence {
///
/// This is a name-and-shape check only; it cannot detect a sabotaged
/// function body. [`verify_floor_guard_behavior`] proves the semantics.
#[datastore_span(name = "replica_fence_verify_catalog", system = "postgresql")]
pub async fn verify_floor_guard_catalog(pool: &PgPool) -> crate::Result<()> {
// tgtype bits: 1 = ROW, 2 = BEFORE, 4 = INSERT, 16 = UPDATE, 64 = INSTEAD.
// Required: ROW + INSERT + UPDATE set, BEFORE + INSTEAD clear.
@@ -369,6 +372,7 @@ pub async fn verify_floor_guard_catalog(pool: &PgPool) -> crate::Result<()> {
/// `SET CONSTRAINTS ALL IMMEDIATE` makes the deferred trigger fire per
/// statement so each adversary is observable under a savepoint; deferral to
/// COMMIT is separately pinned by the held-transaction fixture.
#[datastore_span(name = "replica_fence_verify_behavior", system = "postgresql")]
pub async fn verify_floor_guard_behavior(pool: &PgPool) -> crate::Result<()> {
use crate::error::DbError;
@@ -698,6 +702,10 @@ pub const AURORA_IDENTITY_FN: &str = "aurora_db_instance_identifier";
/// (undefined_function, SQLSTATE 42883); transient errors surface as `Err`
/// so the caller can retry the probe on a later request instead of caching
/// a wrong answer.
#[datastore_span(
name = "replica_fence_reader_supports_aurora_identity",
system = "postgresql"
)]
pub async fn reader_supports_aurora_identity(conn: &mut PgConnection) -> Result<bool, sqlx::Error> {
match sqlx::query(sqlx::AssertSqlSafe(format!(
"SELECT {AURORA_IDENTITY_FN}()"
+1
View File
@@ -19,6 +19,7 @@ path = "src/main.rs"
buzz-core = { workspace = true }
buzz-conformance = { workspace = true }
buzz-db = { workspace = true }
buzz-datastore-tracing = { workspace = true }
buzz-auth = { workspace = true }
buzz-pubsub = { workspace = true }
buzz-audit = { workspace = true }
@@ -19,6 +19,7 @@ use uuid::Uuid;
use buzz_core::kind::*;
use buzz_core::tenant::{CommunityId, TenantContext};
use buzz_datastore_tracing::datastore_span;
use buzz_db::workflow::{ApprovalStatus, RunStatus};
use buzz_db::DbError;
use buzz_workflow::executor::TriggerContext;
@@ -97,6 +98,7 @@ enum PersistResult {
/// persists without the event record. On retry, the event INSERT succeeds
/// (no conflict), and the mutation re-executes — which is safe for idempotent
/// operations (open_dm, hide_dm, update_approval, upsert_workflow).
#[datastore_span(name = "persist_command_event", system = "postgresql")]
async fn persist_command_event(
state: &Arc<AppState>,
tenant: &TenantContext,
+2
View File
@@ -9,9 +9,11 @@ description = "Postgres full-text search for Buzz, scoped by community"
[dependencies]
buzz-core = { workspace = true }
buzz-datastore-tracing = { workspace = true }
sqlx = { workspace = true }
uuid = { workspace = true }
thiserror = { workspace = true }
tracing = { workspace = true }
[dev-dependencies]
tokio = { workspace = true }
+4 -1
View File
@@ -8,10 +8,12 @@
//!
//! See conformance row 50.
use buzz_core::CommunityId;
use sqlx::{PgPool, QueryBuilder, Row};
use uuid::Uuid;
use buzz_core::CommunityId;
use buzz_datastore_tracing::datastore_span;
use crate::error::SearchError;
/// Channel-scope filter for a community-scoped FTS query.
@@ -213,6 +215,7 @@ fn normalized_search_text(q: &str) -> Option<String> {
///
/// `community_id = $ctx` is the first predicate and is non-negotiable. There
/// is no code path through this function that omits it.
#[datastore_span(name = "search", system = "postgresql")]
pub async fn search(pool: &PgPool, query: &SearchQuery) -> Result<SearchResult, SearchError> {
let Some(search_text) = normalized_search_text(&query.q) else {
return Ok(SearchResult {
-78
View File
@@ -1,78 +0,0 @@
# Corporate identity
Corporate identity is an optional relay policy enabled with
`BUZZ_REQUIRE_CORPORATE_IDENTITY=true`. The relay verifies an asymmetric JWT
after the request proves control of a Nostr key, then admits the request only
when the existing community policy also succeeds.
## Required JWT policy
- `BUZZ_CORPORATE_IDENTITY_JWKS_URI` must be HTTPS and contain no credentials.
- JWTs must have a supported asymmetric algorithm, a `kid`, and valid `exp`,
`iss`, and `aud` claims. A present `nbf` claim is enforced.
- `BUZZ_CORPORATE_IDENTITY_NPUB_CLAIM`, when configured, is mandatory and must
equal the authenticated Nostr key. Leaving it unset enables first-use
uid-to-key enrollment in the private binding table.
- JWKS requests have connect and total timeouts, reject redirects, cap the
response at 1 MiB, cache keys for five minutes, and coalesce refreshes.
`BUZZ_REQUIRE_CORPORATE_IDENTITY` and
`BUZZ_ALLOW_CORPORATE_IDENTITY_DELEGATION` are strict booleans. Misspellings and
non-UTF-8 values stop configuration loading instead of silently disabling a
gate.
## Binding and revocation lifecycle
JWT validation is read-only. The relay creates or refreshes a binding only
after admission, allowlist, role, and community membership checks succeed.
Invite claims commit the binding, membership, policy evidence, and invite use
in one PostgreSQL transaction.
Revocation has three explicit meanings:
- `principal` disables every key for an issuer-qualified uid. Normal
authentication cannot re-enroll the principal with another key.
- `key` revokes one key but does not silently authorize a replacement.
- `rotation` is the audit state written by an explicit atomic old-key to
new-key rotation.
WebSocket and audio sessions revalidate the authoritative binding at least
every 30 seconds. Direct sessions also close at JWT expiry. Delegated sessions
check the owner's binding, so disabling an owner evicts the owner's agents as
well as the direct owner session.
Corporate NIP-OA delegation is transport-wide and therefore accepts only an
empty conditions string. Conditional tags must be evaluated for a specific
operation and are not treated as blanket corporate identity authority.
## Privacy and public assertions
`BUZZ_CORPORATE_IDENTITY_DISPLAY_CLAIM` is private. Its default (`email`) is
stored only in the community-scoped binding table and audit data; it is not
published to Nostr.
Public projection is separately opt-in with
`BUZZ_CORPORATE_IDENTITY_PUBLIC_DISPLAY_CLAIM`. When set, that claim is
published as a relay-signed NIP-85 label. Assertions carry both `active=true`
and an `expiration` no later than one hour or the JWT's expiry, whichever comes
first. Clients require the relay signature, active status, and a future
expiration. Removing the opt-in publishes an inactive replacement only when a
prior public assertion exists.
NIP-85 events are replaceable relay events and may also remain in downstream
caches or archives after replacement. Operators must choose a non-sensitive,
user-approved public label and account for that retention when configuring the
public claim.
## Route policy
Corporate identity applies to authenticated WebSocket and audio connections,
the NIP-98 event/query/count bridge, moderation reads, invite mint and claim,
Git smart HTTP, media uploads, and protected media reads.
Intentional exemptions are public media reads when media GET authentication is
disabled, health/readiness/metrics endpoints, NIP-11 and NIP-05 discovery,
operator and admin control planes with their own authentication, secret-backed
workflow hooks, public join-policy documents, invite policy-acceptance
callbacks, and static local web callbacks. These exemptions must remain in the
central route-policy test matrix when routes change.