mirror of
https://github.com/block/buzz.git
synced 2026-10-27 04:30:52 +00:00
## Why
Buzz already reports coarse writer/reader database checkout waits, but
those
signals cannot explain which startup or serving operation is blocked by
pool
pressure. That makes rollout diagnosis and postmortems ambiguous: a
readiness
probe, NIP-42 authentication, authorization check, reconnect history
repair,
event write, and background maintenance can all wait on the same pool
while
appearing identical.
This PR implements Package 2A of the pod-handoff plan: operation-aware
pool
borrow causality. It is observability-only; it does not change
configured pool
sizes, SQL semantics, transaction ordering, or timeout policy. Physical
DNS/TCP/TLS/Postgres authentication and session initialization remain
the
separate Package 2B boundary.
## Metric contract
The final contract separates three questions:
| Question | Metric |
|---|---|
| How long did checkout wait? |
`buzz_db_pool_acquire_duration_seconds{pool_role,operation}` |
| How did the attempt end? |
`buzz_db_pool_acquire_attempts_total{pool_role,operation,outcome}` |
| Who is waiting now for a tracked operation? |
`buzz_db_pool_waiters{pool_role,operation}` |
Outcomes are `success`, `timeout`, `error`, and `cancelled`. Operations
are
`bootstrap`, `readiness`, `tenant_resolution`, `authentication`,
`authorization`, `subscription_history`, `event_write`, and
`maintenance`.
Only these eleven pool/operation pairs are constructible:
```text
writer/bootstrap reader/bootstrap
writer/readiness
writer/tenant_resolution
writer/authentication
writer/authorization reader/authorization
writer/subscription_history reader/subscription_history
writer/event_write
writer/maintenance
```
The duration histogram intentionally does not carry `outcome`. Result
data
remains available on the terminal counter for historical counts and
rates,
without multiplying the expensive histogram family. Nine finite buckets
plus
`+Inf`, sum, and count produce 12 duration series per valid pair.
Together with
four outcome counters and one waiter gauge, the new-family ceiling is
exactly
187 raw Prometheus series per pod, asserted from the production
exporter.
The existing coarse acquisition families remain temporarily for
dashboard
compatibility while the new series are validated in staging.
The operation-specific waiter family covers the explicitly routed,
deployment-critical operations above; it is not a census of every
possible
SQLx checkout in the process. The dashboard pairs it with SQLx pool
active/idle/max gauges for whole-pool capacity context, and treats a
missing
operation series as unknown rather than healthy zero.
## What changed
### Cancellation-safe acquisition ownership
- Add writer- and reader-specific typed operation APIs so invalid label
pairs
cannot be constructed and store modules cannot emit reader labels.
- Own every polled acquisition with one RAII terminal guard.
- Record exactly one duration and terminal outcome for success, timeout,
error,
or cancellation.
- Emit nothing for a future that is created but never polled.
- Balance the operation-specific waiter count exactly once on every
terminal or
dropped future.
- Periodically refresh every expected waiter pair, including healthy
zero, so
missing telemetry is not presented as zero. Reader pairs are emitted
only
when a distinct read pool is configured; a writer-only pod cannot
fabricate
healthy reader-zero state.
### Production attribution
Route the deployment-critical acquisition paths through caller-owned
semantic
entry points, including:
- writer and reader bootstrap;
- the real post-#7149 readiness acquisition and deletion-catalog
validation;
- tenant resolution and community lifecycle checks;
- NIP-42 allowlist authentication;
- membership, moderation, invite, operator, Git, agent-owner, and policy
authorization;
- operator community create/list/archive/unarchive, reverse host/channel
tenant
resolution, and REQ row-community conformance lookups;
- writer/reader subscription history, feed, thread, and routed fallback
paths;
- primary and command event writes, replaceable events, mention
indexing,
reaction/channel/member/archive side effects, and thread metadata;
- push matching, usage rollups/leadership, replica-fence startup and
recurring
probes, periodic reconciliation, channel/deletion reapers, partitions,
and
other bounded maintenance/bootstrap paths.
Shared helpers now accept caller-owned intent or expose named semantic
variants
instead of assigning one misleading operation to every caller. No known
P0 path
uses `other`.
### Readiness and size-one-pool correctness
- Rebase on the post-#7149 readiness implementation and instrument the
actual
`Db::readiness_check` acquisition rather than the superseded ping-only
seam.
- Acquire once for deletion-catalog validation and run its queries on
that
connection, preserving the shared readiness deadline.
- Scope the channel-roster catalog checkout before the behavior probe so
a
size-one writer pool cannot self-deadlock during startup verification.
### Exporter, documentation, and CI
- Register metric HELP/type/unit metadata through the production
Prometheus
builder.
- Configure dedicated checkout buckets at 1ms, 5ms, 10ms, 25ms, 50ms,
150ms,
500ms, 1s, and 3s.
- Add a production scrape-contract test for exact names, labels,
buckets,
valid pairs, sensitive-label exclusion, and the 187-series ceiling.
- Add source mutation guards for the P0 semantic entry points and
raw-checkout
bypasses.
- Add an exact backend-integration CI selector for the production
attribution,
cancellation, readiness, and size-one-pool PostgreSQL tests.
- Document the frozen label vocabulary, valid combinations, semantics,
and
cardinality budget in the Helm chart README.
## Dashboard intent
The new Stage 2 row in **Buzz Startup & Rollout Safety** is
deployment-first:
- baseline-versus-candidate attempts, failure rates, cancellation rates,
and
maximum wait by operation;
- outcome counts and percentages over time by SHA/ReplicaSet;
- acquisition wait heatmap, average, and maximum through the rollout;
- historical waiter pressure beside writer active/idle/max context;
- per-pod postmortem drilldown, including terminated pods;
- a smaller current-waiter table with explicit stale/missing semantics.
Percentile widgets remain disabled until Datadog metadata confirms
percentile
support for the new distribution. Current gauges use no fill,
interpolation, or
`default_zero`; missing means unknown.
## Risk assessment
Moderate. The patch touches many database acquisition call sites, but
preserves
the selected physical pool and executes the same SQL on the acquired
connection. The main risks are incorrect semantic attribution,
cancellation
double-counting, and a helper accidentally acquiring twice. Typed APIs,
production-method PostgreSQL tests, source guards, the raw scrape
contract, and
the size-one-pool regression cover those risks.
No tenant, community, user, pubkey, event, channel, SQL, URL, pod,
version,
ReplicaSet, or request-controlled value is emitted as an application
metric
label. Deployment identity is supplied by infrastructure enrichment.
## Verification
- `cargo fmt --all -- --check` — passed.
- `cargo clippy -p buzz-db -p buzz-relay --all-targets --all-features --
-D warnings`
— passed.
- `cargo test -p buzz-db` — 122 passed, 0 failed, 263 ignored;
source-contract integration test: 3 passed, 0 failed.
- Focused relay compatibility, metric-contract, and readiness tests —
passed.
- `scripts/test-postgres-test-discovery.sh` — passed.
- Full `buzz-relay` package run from the identical tree reached 1,015
passes;
the six media-test failures all stopped in their shared local PostgreSQL
setup with `Sqlx(PoolTimedOut)` because Docker/PostgreSQL was
unavailable.
The same six failed in isolation, while every changed exact test passed.
- Exact implementation head:
`f92910b353086e9edf85918ca5f72190edbbe22f`.
- Exact multi-architecture staging image:
`dev-sha-f92910b353086e9edf85918ca5f72190edbbe22f-run-33607968668-1`
(`sha256:161712c8ed2e265a15df9b63e02248d5973481f875ff129d7d2ae78a09d487a2`).
- Focused staging GitOps PR:
<https://github.com/squareup/builderbot-platform-core-infrastructure/pull/299>
— merged after renderer, inventory, infrastructure test, Kargo, Semgrep,
and
Intersect gates passed; the source/generated-artifact diff was exactly
two
image lines.
- Exact GitHub head reports 47 terminal checks: 35 successful and 12
intentionally skipped. PostgreSQL, unit, lint, security, both server
cross-compiles, backend integration, relay E2E, desktop, mobile, image,
Helm, Semgrep, zizmor, and DCO gates are green.
- Datadog readback identifies two exact-image pods,
`buzz-d79c8d8f7-ckv2l` and `buzz-d79c8d8f7-qzqdp`, in ReplicaSet
`buzz-d79c8d8f7`; both report the full source SHA above.
- Both pods report all eleven allowed pool/operation waiter pairs at
current
zero, with no invalid pair. The acceptance window observed nonzero
success
receipts for readiness, tenant resolution, authorization, subscription
history, event write, and maintenance, and no timeout, error, or
cancelled
outcome. Maximum observed wait was about 101 ms for maintenance and 50
ms
for reader subscription history.
The main **Buzz Startup & Rollout Safety** dashboard now has a live
Stage 2
database row with eight widgets and nineteen fully scoped queries. Final
readback preserved all seven top-level groups, found zero under-scoped
Row 6
queries, and confirmed the tracked-operation waiter boundary in the
panel
descriptions.
Generated with Codex.
---------
Signed-off-by: Ravneet Arora <rarora@squareup.com>
628 lines
27 KiB
Rust
628 lines
27 KiB
Rust
#[test]
|
|
fn database_metrics_and_slow_logs_exclude_sensitive_or_unbounded_fields() {
|
|
let implementation = include_str!("../src/runtime/observability.rs");
|
|
let datastore_macro = include_str!("../../buzz-datastore-tracing/src/lib.rs");
|
|
let instrumentation = format!("{implementation}\n{datastore_macro}");
|
|
|
|
for forbidden in [
|
|
"\"community\" =>",
|
|
"\"event_id\" =>",
|
|
"\"event_kind\" =>",
|
|
"\"kind\" =>",
|
|
"\"sql\" =>",
|
|
"\"query\" =>",
|
|
"\"query_id\" =>",
|
|
"\"d_tag\" =>",
|
|
"\"coordinate\" =>",
|
|
"community =",
|
|
"event_id =",
|
|
"event_kind =",
|
|
"sql =",
|
|
"query_id =",
|
|
"d_tag =",
|
|
"coordinate =",
|
|
] {
|
|
assert!(
|
|
!instrumentation.contains(forbidden),
|
|
"database instrumentation must not expose {forbidden}"
|
|
);
|
|
}
|
|
|
|
assert!(datastore_macro.contains("name: LitStr"));
|
|
assert!(datastore_macro.contains("\"operation\" => #name"));
|
|
assert!(datastore_macro.contains("elapsed_ms ="));
|
|
assert!(
|
|
datastore_macro.contains("parent: None"),
|
|
"slow warnings must not inherit dynamic datastore span fields"
|
|
);
|
|
// The runtime tracing-layer assertion covers field names because a source
|
|
// search would also match ordinary local variables such as `record_error`.
|
|
}
|
|
|
|
#[test]
|
|
fn relay_admin_db_wrappers_have_exactly_one_datastore_span() {
|
|
for (domain, source) in [
|
|
(
|
|
"relay_admin_actions",
|
|
include_str!("../src/store/relay_admin_actions.rs"),
|
|
),
|
|
(
|
|
"relay_operators",
|
|
include_str!("../src/store/relay_operators.rs"),
|
|
),
|
|
] {
|
|
let db_impl = source
|
|
.split_once("impl crate::Db {")
|
|
.unwrap_or_else(|| panic!("{domain} must own its Db wrappers"))
|
|
.1
|
|
.split_once("\n#[cfg(test)]")
|
|
.unwrap_or_else(|| panic!("{domain} Db wrappers must precede focused tests"))
|
|
.0;
|
|
let mut pending_spans = 0;
|
|
let mut methods = 0;
|
|
|
|
for line in db_impl.lines() {
|
|
if line.contains("#[datastore_span(") {
|
|
pending_spans += 1;
|
|
}
|
|
if line.trim_start().starts_with("pub async fn ") {
|
|
assert_eq!(pending_spans, 1, "{domain} wrapper `{line}` span count");
|
|
pending_spans = 0;
|
|
methods += 1;
|
|
}
|
|
}
|
|
|
|
assert!(methods > 0, "{domain} must own public Db wrappers");
|
|
assert_eq!(
|
|
pending_spans, 0,
|
|
"{domain} has an unattached datastore span"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn p0_pool_acquisitions_use_typed_operation_pairs_without_other() {
|
|
let observability = include_str!("../src/runtime/observability.rs");
|
|
assert!(observability.contains("enum PoolOperation"));
|
|
assert!(observability.contains("pub(crate) enum WriterOperation"));
|
|
assert!(observability.contains("pub(crate) enum ReaderOperation"));
|
|
assert!(observability.contains("Self::WriterAuthentication"));
|
|
assert!(observability.contains("Self::ReaderSubscriptionHistory"));
|
|
assert!(observability.contains("pub(crate) async fn acquire_writer("));
|
|
assert!(observability.contains("pub(super) async fn acquire_reader_with_legacy_metrics("));
|
|
assert!(observability.contains("static POOL_WAITERS: [Mutex<u64>"));
|
|
assert!(!observability.contains("AtomicU64"));
|
|
assert!(!observability.contains("DbOperation::Other"));
|
|
assert!(!observability.contains("\"other\""));
|
|
assert!(!observability.contains("buzz_db_pool_acquire_timeouts_total"));
|
|
assert!(!observability.contains("\"result\" =>"));
|
|
let legacy_transaction = observability
|
|
.split_once("pub(crate) async fn begin_transaction(")
|
|
.expect("observability must expose attributed transaction acquisition")
|
|
.1
|
|
.split_once("pub(crate) async fn observe_advisory_lock")
|
|
.expect("transaction acquisition must precede advisory-lock observation")
|
|
.0;
|
|
assert!(legacy_transaction.contains("acquire_writer_with_legacy_metrics("));
|
|
|
|
let runtime = include_str!("../src/runtime/mod.rs");
|
|
assert!(runtime.contains("observability::acquire_writer_until("));
|
|
assert!(runtime.contains("WriterOperation::Readiness"));
|
|
assert!(runtime.contains("WriterOperation::EventWrite"));
|
|
assert!(runtime.contains("ReaderOperation::Bootstrap"));
|
|
assert!(runtime.contains("pub async fn begin_event_write_transaction"));
|
|
let reader_boot = runtime
|
|
.split_once("async fn read_pool_boot_ping_once(")
|
|
.expect("runtime must expose the reader boot probe")
|
|
.1
|
|
.split_once("#[cfg(test)]")
|
|
.expect("reader boot probe must precede its test seam")
|
|
.0;
|
|
assert!(reader_boot.contains("acquire_reader_with_legacy_metrics("));
|
|
let routed_reader = runtime
|
|
.split_once("async fn proved_reader(")
|
|
.expect("runtime must expose the routed-reader checkout")
|
|
.1
|
|
.split_once("async fn reader_aurora_capability_on(")
|
|
.expect("routed-reader checkout must precede capability probing")
|
|
.0;
|
|
assert!(routed_reader.contains("acquire_reader_with_legacy_metrics(read_pool, operation)"));
|
|
let event_write_transaction = runtime
|
|
.split_once("pub async fn begin_event_write_transaction(")
|
|
.expect("runtime must expose the legacy event-write transaction seam")
|
|
.1
|
|
.split_once("pub async fn insert_event_with_serving_write_guard(")
|
|
.expect("legacy event-write transaction must precede guarded writes")
|
|
.0;
|
|
assert!(event_write_transaction.contains("acquire_writer_with_legacy_metrics("));
|
|
|
|
let migration = include_str!("../src/runtime/migration.rs");
|
|
let migration_lock = migration
|
|
.split_once("pub(crate) async fn with_exclusive_schema_destruction_lock")
|
|
.expect("migration must expose the schema-safety acquisition seam")
|
|
.1
|
|
.split_once("async fn reject_legacy_nip_rs_cardinality_ambiguity")
|
|
.expect("schema-safety acquisition must precede migration validation")
|
|
.0;
|
|
assert!(migration_lock.contains("acquire_writer_with_legacy_metrics("));
|
|
|
|
let allowlist = include_str!("../src/store/allowlist.rs");
|
|
assert!(allowlist.contains("WriterOperation::Authentication"));
|
|
assert!(allowlist.contains("WriterOperation::Authorization"));
|
|
assert!(!allowlist.contains("fetch_one(&self.pool)"));
|
|
|
|
let event = include_str!("../src/store/event.rs");
|
|
assert!(event.contains("query_events_with_operation"));
|
|
assert!(event.contains("WriterOperation::Authorization"));
|
|
assert!(event.contains("WriterOperation::SubscriptionHistory"));
|
|
assert!(event.contains("ReaderOperation::SubscriptionHistory"));
|
|
let backfill_d_tags = event
|
|
.split_once("pub async fn backfill_d_tags")
|
|
.expect("event store must expose the startup d-tag backfill")
|
|
.1
|
|
.split_once("/// Soft-delete NIP-29 discovery events")
|
|
.expect("d-tag backfill must precede discovery deletion")
|
|
.0;
|
|
assert!(backfill_d_tags.contains("WriterOperation::Bootstrap"));
|
|
assert!(backfill_d_tags.contains("execute(&mut *connection)"));
|
|
let soft_delete_discovery = event
|
|
.split_once("pub async fn soft_delete_discovery_events")
|
|
.expect("event store must expose discovery-event deletion")
|
|
.1
|
|
.split_once("\n}\n\n#[cfg(test)]")
|
|
.expect("discovery deletion must end the production Db implementation")
|
|
.0;
|
|
assert!(soft_delete_discovery.contains("WriterOperation::EventWrite"));
|
|
assert!(soft_delete_discovery.contains("execute(&mut *connection)"));
|
|
|
|
let side_effects = include_str!("../../buzz-relay/src/handlers/side_effects.rs");
|
|
assert!(side_effects.contains("query_events_for_event_write"));
|
|
assert!(side_effects.contains("query_events_for_bootstrap"));
|
|
assert!(side_effects.contains(".list_channels_for_bootstrap("));
|
|
|
|
let deletion = include_str!("../src/store/deletion.rs");
|
|
let public_serving_catalog = deletion
|
|
.split_once("pub async fn validate_serving_catalog(&self)")
|
|
.expect("deletion store must preserve its public serving-catalog API")
|
|
.1
|
|
.split_once("async fn validate_serving_catalog_on")
|
|
.expect("public serving-catalog validation must delegate to its connection helper")
|
|
.0;
|
|
assert!(public_serving_catalog.contains("WriterOperation::Bootstrap"));
|
|
assert!(public_serving_catalog.contains("observability::acquire_writer("));
|
|
assert!(public_serving_catalog.contains("validate_serving_catalog_on"));
|
|
assert!(!public_serving_catalog.contains("self.pool.acquire()"));
|
|
|
|
let thread = include_str!("../src/store/thread.rs");
|
|
let thread_metadata = thread
|
|
.split_once("pub async fn get_thread_metadata_by_event(")
|
|
.expect("thread store must expose metadata lookup")
|
|
.1
|
|
.split_once("// -- Db API")
|
|
.expect("metadata lookup must precede the Db wrapper section")
|
|
.0;
|
|
assert!(thread_metadata.contains("WriterOperation::EventWrite"));
|
|
assert!(thread_metadata.contains("fetch_optional(&mut *connection)"));
|
|
assert!(!thread_metadata.contains("fetch_optional(pool)"));
|
|
|
|
let channel = include_str!("../src/store/channel.rs");
|
|
assert!(channel.contains("async fn begin_event_write_transaction("));
|
|
assert!(channel.contains("async fn acquire_event_write_connection("));
|
|
for (start, end, expected) in [
|
|
(
|
|
"pub async fn create_channel(\n",
|
|
"/// Creates a channel with a client-supplied UUID",
|
|
"begin_event_write_transaction(pool)",
|
|
),
|
|
(
|
|
"pub async fn create_channel_with_id(\n",
|
|
"/// Fetches a channel record by `(community_id, id)`",
|
|
"begin_event_write_transaction(pool)",
|
|
),
|
|
(
|
|
"pub async fn update_channel(\n",
|
|
"/// Sets the topic for a channel",
|
|
"begin_event_write_transaction(pool)",
|
|
),
|
|
(
|
|
"pub async fn set_topic(\n",
|
|
"/// Sets the purpose for a channel",
|
|
"acquire_event_write_connection(pool)",
|
|
),
|
|
(
|
|
"pub async fn set_purpose(\n",
|
|
"/// Archives a channel",
|
|
"acquire_event_write_connection(pool)",
|
|
),
|
|
(
|
|
"pub async fn archive_channel(\n",
|
|
"/// Unarchives a channel",
|
|
"acquire_event_write_connection(pool)",
|
|
),
|
|
(
|
|
"pub async fn unarchive_channel(\n",
|
|
"/// Soft-delete a channel",
|
|
"acquire_event_write_connection(pool)",
|
|
),
|
|
(
|
|
"pub async fn soft_delete_channel(\n",
|
|
"/// Archive ephemeral channels",
|
|
"acquire_event_write_connection(pool)",
|
|
),
|
|
] {
|
|
let function = channel
|
|
.split_once(start)
|
|
.unwrap_or_else(|| panic!("missing channel seam {start}"))
|
|
.1
|
|
.split_once(end)
|
|
.unwrap_or_else(|| panic!("channel seam {start} must precede {end}"))
|
|
.0;
|
|
assert!(
|
|
function.contains(expected),
|
|
"channel seam {start} must use {expected}"
|
|
);
|
|
assert!(!function.contains("pool.begin().await"));
|
|
assert!(!function.contains(".execute(pool)"));
|
|
assert!(!function.contains(".fetch_optional(pool)"));
|
|
}
|
|
let get_channel = channel
|
|
.split_once("async fn get_channel_with_operation(")
|
|
.expect("channel store must route shared lookups through caller-owned intent")
|
|
.1
|
|
.split_once("/// Returns the canvas content")
|
|
.expect("channel lookup helper must precede canvas reads")
|
|
.0;
|
|
assert!(get_channel.contains("acquire_writer(pool, operation)"));
|
|
assert!(get_channel.contains("fetch_optional(&mut *connection)"));
|
|
assert!(!get_channel.contains("fetch_optional(pool)"));
|
|
assert!(channel.contains("pub async fn get_channel_for_event_write("));
|
|
let list_channels = channel
|
|
.split_once("async fn list_channels_with_operation(")
|
|
.expect("channel listing must accept caller-owned intent")
|
|
.1
|
|
.split_once("/// A channel archived by the ephemeral-channel reaper")
|
|
.expect("channel listing must precede ephemeral-channel types")
|
|
.0;
|
|
assert!(list_channels.contains("acquire_writer(pool, operation)"));
|
|
assert!(list_channels.contains("fetch_all(&mut *connection)"));
|
|
assert!(!list_channels.contains("fetch_all(pool)"));
|
|
assert!(channel.contains("pub async fn list_channels_for_bootstrap("));
|
|
|
|
let channel_members = include_str!("../src/store/channel_members.rs");
|
|
assert!(channel_members.contains("async fn get_members_with_operation("));
|
|
assert!(channel_members.contains("pub async fn get_members_for_event_write("));
|
|
assert!(channel_members.contains("async fn get_users_bulk_with_operation("));
|
|
assert!(channel_members.contains("pub async fn get_users_bulk_for_event_write("));
|
|
|
|
let huddle_link = event
|
|
.split_once("async fn huddle_started_link_exists_with_operation(")
|
|
.expect("huddle link lookup must accept caller-owned intent")
|
|
.1
|
|
.split_once("/// Insert a Nostr event")
|
|
.expect("huddle link lookup must precede event insertion")
|
|
.0;
|
|
assert!(huddle_link.contains("acquire_writer(pool, operation)"));
|
|
assert!(event.contains("pub async fn huddle_started_link_exists_for_event_write("));
|
|
let ingest = include_str!("../../buzz-relay/src/handlers/ingest.rs");
|
|
assert!(ingest.contains(".huddle_started_link_exists_for_event_write("));
|
|
let audio = include_str!("../../buzz-relay/src/audio/handler.rs");
|
|
assert!(audio.contains(".huddle_started_link_exists("));
|
|
|
|
let workflow_sink = include_str!("../../buzz-relay/src/workflow_sink.rs");
|
|
assert!(workflow_sink.contains(".get_members_for_event_write("));
|
|
assert!(workflow_sink.contains(".get_users_bulk_for_event_write("));
|
|
|
|
for write_caller in [
|
|
include_str!("../../buzz-relay/src/handlers/side_effects.rs"),
|
|
include_str!("../../buzz-relay/src/handlers/ingest.rs"),
|
|
include_str!("../../buzz-relay/src/handlers/command_executor.rs"),
|
|
workflow_sink,
|
|
] {
|
|
assert!(!write_caller.contains(".get_channel("));
|
|
assert!(write_caller.contains(".get_channel_for_event_write("));
|
|
}
|
|
|
|
let user = include_str!("../src/store/user.rs");
|
|
let agent_channel_policy = user
|
|
.split_once("pub async fn get_agent_channel_policy(")
|
|
.expect("user store must expose get_agent_channel_policy")
|
|
.1
|
|
.split_once("/// Check whether `actor_pubkey`")
|
|
.expect("agent policy lookup must precede owner lookup")
|
|
.0;
|
|
assert!(agent_channel_policy.contains("WriterOperation::Authorization"));
|
|
assert!(agent_channel_policy.contains("fetch_optional(&mut *connection)"));
|
|
assert!(!agent_channel_policy.contains("fetch_optional(pool)"));
|
|
let is_agent_owner = user
|
|
.split_once("pub async fn is_agent_owner(")
|
|
.expect("user store must expose is_agent_owner")
|
|
.1
|
|
.split_once("/// Set the channel_add_policy")
|
|
.expect("is_agent_owner must precede set_agent_channel_policy")
|
|
.0;
|
|
assert!(is_agent_owner.contains("WriterOperation::Authorization"));
|
|
assert!(is_agent_owner.contains("acquire_writer("));
|
|
assert!(is_agent_owner.contains("fetch_optional(&mut *connection)"));
|
|
assert!(!is_agent_owner.contains("fetch_optional(pool)"));
|
|
|
|
let moderation = include_str!("../src/store/moderation.rs");
|
|
let restriction_state = moderation
|
|
.split_once("pub async fn restriction_state(")
|
|
.expect("moderation store must expose restriction_state")
|
|
.1
|
|
.split_once("/// Fetch the full ban/timeout row")
|
|
.expect("restriction state must precede full ban reads")
|
|
.0;
|
|
assert!(restriction_state.contains("WriterOperation::Authorization"));
|
|
assert!(restriction_state.contains("fetch_optional(&mut *connection)"));
|
|
assert!(!restriction_state.contains("fetch_optional(pool)"));
|
|
|
|
let community_store = include_str!("../src/store/community.rs");
|
|
let ensure_community = community_store
|
|
.split_once("pub async fn ensure_configured_community(")
|
|
.expect("community store must expose ensure_configured_community")
|
|
.1
|
|
.split_once("/// Atomically creates a community")
|
|
.expect("configured-community helpers must precede community creation")
|
|
.0;
|
|
assert!(ensure_community.contains("WriterOperation::Authorization"));
|
|
assert!(ensure_community.contains("WriterOperation::Bootstrap"));
|
|
assert!(ensure_community.contains("ensure_configured_community_with_operation"));
|
|
assert!(ensure_community.contains("acquire_writer(&self.pool, operation)"));
|
|
assert!(ensure_community.contains("fetch_optional(&mut *connection)"));
|
|
let management_lookup = community_store
|
|
.split_once("pub async fn lookup_community_by_host_for_management(")
|
|
.expect("community store must expose management host lookup")
|
|
.1
|
|
.split_once("/// Lists communities where")
|
|
.expect("management lookup must precede owner listing")
|
|
.0;
|
|
assert!(management_lookup.contains("WriterOperation::Authorization"));
|
|
assert!(management_lookup.contains("fetch_optional(&mut *connection)"));
|
|
assert!(!management_lookup.contains("fetch_optional(&self.pool)"));
|
|
let community_production = community_store
|
|
.split("\n#[cfg(test)]")
|
|
.next()
|
|
.expect("community production source");
|
|
for required in [
|
|
"WriterOperation::TenantResolution",
|
|
"WriterOperation::Authorization",
|
|
"WriterOperation::SubscriptionHistory",
|
|
"WriterOperation::EventWrite",
|
|
] {
|
|
assert!(
|
|
community_production.contains(required),
|
|
"community P0 paths must include {required} attribution"
|
|
);
|
|
}
|
|
assert!(!community_production.contains("self.pool.begin().await"));
|
|
assert!(!community_production.contains(".fetch_one(&self.pool)"));
|
|
assert!(!community_production.contains(".fetch_all(&self.pool)"));
|
|
assert!(!community_production.contains(".execute(&self.pool)"));
|
|
assert_eq!(
|
|
community_production
|
|
.matches(".fetch_optional(&self.pool)")
|
|
.count(),
|
|
1,
|
|
"only the out-of-scope NIP-11 metadata read may retain a raw pool checkout"
|
|
);
|
|
|
|
let thread_summary = thread
|
|
.split_once("pub async fn get_thread_summary(")
|
|
.expect("thread store must expose get_thread_summary")
|
|
.1
|
|
.split_once("/// Fetch one channel window")
|
|
.expect("thread summary must precede channel-window reads")
|
|
.0;
|
|
assert!(thread_summary.contains("WriterOperation::EventWrite"));
|
|
assert!(thread_summary.contains("fetch_optional(&mut *connection)"));
|
|
assert!(thread_summary.contains("fetch_all(&mut *connection)"));
|
|
assert!(!thread_summary.contains("fetch_optional(pool)"));
|
|
assert!(!thread_summary.contains("fetch_all(pool)"));
|
|
|
|
let archived_identities = include_str!("../src/store/archived_identities.rs");
|
|
let archived_identity_production = archived_identities
|
|
.split("\n#[cfg(test)]")
|
|
.next()
|
|
.expect("archived identity production source");
|
|
assert_eq!(
|
|
archived_identity_production
|
|
.matches("WriterOperation::EventWrite")
|
|
.count(),
|
|
4,
|
|
"all four archived identity operations must be attributed to event writes"
|
|
);
|
|
assert!(!archived_identity_production.contains("fetch_optional(pool)"));
|
|
assert!(!archived_identity_production.contains("fetch_all(pool)"));
|
|
assert!(!archived_identity_production.contains("execute(pool)"));
|
|
|
|
let relay_main = include_str!("../../buzz-relay/src/main.rs");
|
|
assert!(relay_main.contains("pool_state.db.refresh_pool_waiter_metrics();"));
|
|
assert!(relay_main.contains(".ensure_configured_community_for_bootstrap("));
|
|
|
|
let runtime = include_str!("../src/runtime/mod.rs");
|
|
assert!(runtime.contains("observability::refresh_pool_waiters(self.read_pool.is_some())"));
|
|
assert!(runtime.contains("self.verify_replica_fence_at_boot().await?"));
|
|
let fence_boot = runtime
|
|
.split_once("pub(crate) async fn verify_replica_fence_at_boot")
|
|
.expect("runtime must expose attributed boot fence verification")
|
|
.1
|
|
.split_once("/// The pool for lag-tolerant reads")
|
|
.expect("boot fence verification must precede routed-read plumbing")
|
|
.0;
|
|
assert!(fence_boot.contains("WriterOperation::Bootstrap"));
|
|
|
|
let replica_fence = include_str!("../src/runtime/replica_fence.rs");
|
|
let replica_fence_production = replica_fence
|
|
.split("\n#[cfg(test)]")
|
|
.next()
|
|
.expect("replica-fence production source");
|
|
assert!(replica_fence_production.contains("WriterOperation::Bootstrap"));
|
|
assert!(replica_fence_production.contains("WriterOperation::Maintenance"));
|
|
assert!(!replica_fence_production.contains("pool.begin().await"));
|
|
assert!(!replica_fence_production.contains("writer.acquire().await"));
|
|
assert!(!replica_fence_production.contains("fetch_optional(writer)"));
|
|
|
|
let usage = include_str!("../src/store/usage.rs");
|
|
let usage_production = usage
|
|
.split("\n#[cfg(test)]")
|
|
.next()
|
|
.expect("usage production source");
|
|
let usage_leader_lock = usage_production
|
|
.split_once("pub async fn try_lock_usage_metrics(")
|
|
.expect("usage store must expose the legacy leader-lock acquisition")
|
|
.1
|
|
.split_once("pub async fn usage_community_count(")
|
|
.expect("usage leader lock must precede counter reads")
|
|
.0;
|
|
assert!(usage_leader_lock.contains("acquire_writer_with_legacy_metrics("));
|
|
assert!(
|
|
usage_production
|
|
.matches("WriterOperation::Maintenance")
|
|
.count()
|
|
>= 11,
|
|
"every periodic usage checkout must be maintenance-attributed"
|
|
);
|
|
for bypass in [
|
|
".fetch_one(pool)",
|
|
".fetch_all(pool)",
|
|
".fetch_optional(pool)",
|
|
".execute(pool)",
|
|
] {
|
|
assert!(
|
|
!usage_production.contains(bypass),
|
|
"usage production path bypasses operation attribution with {bypass}"
|
|
);
|
|
}
|
|
|
|
let channel_reaper = channel
|
|
.split_once("pub async fn reap_expired_ephemeral_channels(pool:")
|
|
.expect("channel store must expose ephemeral reaper")
|
|
.1
|
|
.split_once("\nimpl Db {")
|
|
.expect("ephemeral reaper must precede Db wrappers")
|
|
.0;
|
|
assert!(channel_reaper.contains("WriterOperation::Maintenance"));
|
|
assert!(channel_reaper.contains("fetch_all(&mut *connection)"));
|
|
|
|
let deletion = include_str!("../src/store/deletion.rs");
|
|
let lease_reaper = deletion
|
|
.split_once("pub async fn reap_expired_serving_write_leases")
|
|
.expect("deletion store must expose serving-lease reaper")
|
|
.1
|
|
.split_once("/// Return serving-lease counts")
|
|
.expect("serving-lease reaper must precede stats")
|
|
.0;
|
|
assert!(lease_reaper.contains("WriterOperation::Maintenance"));
|
|
assert!(lease_reaper.contains("execute(&mut *connection)"));
|
|
let lease_stats = deletion
|
|
.split_once("pub async fn serving_lease_stats")
|
|
.expect("deletion store must expose serving-lease stats")
|
|
.1
|
|
.split_once("/// Whether a community remains active")
|
|
.expect("serving-lease stats must precede serving-state reads")
|
|
.0;
|
|
assert!(lease_stats.contains("WriterOperation::Maintenance"));
|
|
assert!(lease_stats.contains("fetch_one(&mut *connection)"));
|
|
for (start, end) in [
|
|
(
|
|
"pub async fn acquire_serving_write_lease",
|
|
"/// Renew an already-admitted external side-effect lease",
|
|
),
|
|
(
|
|
"pub async fn renew_serving_write_lease",
|
|
"/// Release a serving side-effect lease",
|
|
),
|
|
(
|
|
"pub async fn release_serving_write_lease",
|
|
"/// Check that an external side-effect lease remains current",
|
|
),
|
|
(
|
|
"pub async fn verify_serving_write_lease",
|
|
"/// Delete expired serving leases",
|
|
),
|
|
(
|
|
"pub async fn is_serving_active",
|
|
"async fn advance_with_checkpoint",
|
|
),
|
|
] {
|
|
let function = deletion
|
|
.split_once(start)
|
|
.unwrap_or_else(|| panic!("missing serving-write seam {start}"))
|
|
.1
|
|
.split_once(end)
|
|
.unwrap_or_else(|| panic!("serving-write seam {start} must precede {end}"))
|
|
.0;
|
|
assert!(
|
|
function.contains("WriterOperation::EventWrite"),
|
|
"serving-write seam {start} must be event-write attributed"
|
|
);
|
|
assert!(!function.contains("self.pool.begin().await"));
|
|
assert!(!function.contains(".execute(&self.pool)"));
|
|
assert!(!function.contains(".fetch_one(&self.pool)"));
|
|
}
|
|
|
|
let ensure_authorization = user
|
|
.split_once("pub async fn ensure_user_for_authorization(")
|
|
.expect("user store must expose NIP-OA authorization ensure")
|
|
.1
|
|
.split_once("/// Get a single user record")
|
|
.expect("authorization ensure must precede generic user reads")
|
|
.0;
|
|
assert!(ensure_authorization.contains("WriterOperation::Authorization"));
|
|
let set_owner_authorization = user
|
|
.split_once("pub async fn set_agent_owner_for_authorization(")
|
|
.expect("user store must expose NIP-OA authorization owner write")
|
|
.1
|
|
.split_once("/// Get the channel_add_policy")
|
|
.expect("authorization owner write must precede policy reads")
|
|
.0;
|
|
assert!(set_owner_authorization.contains("WriterOperation::Authorization"));
|
|
let relay_api = include_str!("../../buzz-relay/src/api/mod.rs");
|
|
assert!(relay_api.contains(".ensure_user_for_authorization("));
|
|
assert!(relay_api.contains(".set_agent_owner_for_authorization("));
|
|
|
|
for (domain, source) in [
|
|
(
|
|
"channel_members",
|
|
include_str!("../src/store/channel_members.rs"),
|
|
),
|
|
("archived_identities", archived_identities),
|
|
("event", event),
|
|
("git_repo", include_str!("../src/store/git_repo.rs")),
|
|
("push", include_str!("../src/store/push.rs")),
|
|
("replica_fence", replica_fence),
|
|
("reaction", include_str!("../src/store/reaction.rs")),
|
|
("relay_invite", include_str!("../src/store/relay_invite.rs")),
|
|
(
|
|
"relay_members",
|
|
include_str!("../src/store/relay_members.rs"),
|
|
),
|
|
("thread", thread),
|
|
(
|
|
"relay_operators",
|
|
include_str!("../src/store/relay_operators.rs"),
|
|
),
|
|
("usage", usage),
|
|
] {
|
|
let production = source.split("\n#[cfg(test)]").next().unwrap_or(source);
|
|
for bypass in [
|
|
"pool.begin().await",
|
|
"self.pool.begin().await",
|
|
".fetch_one(pool)",
|
|
".fetch_one(&self.pool)",
|
|
".fetch_all(pool)",
|
|
".fetch_all(&self.pool)",
|
|
".fetch_optional(pool)",
|
|
".fetch_optional(&self.pool)",
|
|
".execute(pool)",
|
|
".execute(&self.pool)",
|
|
] {
|
|
assert!(
|
|
!production.contains(bypass),
|
|
"{domain} production path bypasses operation attribution with {bypass}"
|
|
);
|
|
}
|
|
}
|
|
}
|