mirror of
https://github.com/block/buzz.git
synced 2026-10-27 04:30:52 +00:00
6e6d290030
The startup no longer probes the roster trigger, so the comment on the roster guard exceptions now names only the replica-floor probe. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Logan Johnson <loganj@squareup.com>
2157 lines
86 KiB
Rust
2157 lines
86 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 public event-write transaction seam")
|
|
.1
|
|
.split_once("pub async fn insert_event_with_serving_write_guard(")
|
|
.expect("public event-write transaction must precede guarded writes")
|
|
.0;
|
|
assert!(event_write_transaction.contains(COMMUNITY_CHOKEPOINT_LEGACY_MARKER));
|
|
|
|
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 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("begin_community_event_write_transaction("));
|
|
assert!(soft_delete_discovery.contains("execute(tx.conn())"));
|
|
|
|
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("begin_community_write_transaction("));
|
|
for (start, end, expected) in [
|
|
(
|
|
"pub async fn create_channel(\n",
|
|
"/// Creates a channel with a client-supplied UUID",
|
|
"begin_community_write_transaction(",
|
|
),
|
|
(
|
|
"pub async fn create_channel_with_id(\n",
|
|
"/// Fetches a channel record by `(community_id, id)`",
|
|
"begin_community_write_transaction(",
|
|
),
|
|
(
|
|
"pub async fn update_channel(\n",
|
|
"/// Sets the topic for a channel",
|
|
"begin_community_write_transaction(",
|
|
),
|
|
(
|
|
"pub async fn set_topic(\n",
|
|
"/// Sets the purpose for a channel",
|
|
"begin_community_write_transaction(",
|
|
),
|
|
(
|
|
"pub async fn set_purpose(\n",
|
|
"/// Archives a channel",
|
|
"begin_community_write_transaction(",
|
|
),
|
|
(
|
|
"pub async fn archive_channel(\n",
|
|
"/// Unarchives a channel",
|
|
"begin_community_write_transaction(",
|
|
),
|
|
(
|
|
"pub async fn unarchive_channel(\n",
|
|
"/// Soft-delete a channel",
|
|
"begin_community_write_transaction(",
|
|
),
|
|
(
|
|
"pub async fn soft_delete_channel(\n",
|
|
"/// Archive ephemeral channels",
|
|
"begin_community_write_transaction(",
|
|
),
|
|
] {
|
|
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"));
|
|
// The single aggregate row is read on the attributed writer connection,
|
|
// never directly on the pool.
|
|
assert!(restriction_state.contains("fetch_one(&mut *connection)"));
|
|
assert!(!restriction_state.contains("(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_deployment_community(&db,"));
|
|
let provisioning = include_str!("../../buzz-relay/src/handlers/community_provisioning.rs");
|
|
assert!(provisioning.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_session_leader(")
|
|
.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)"));
|
|
// Lease seams that delegate to the bounded helper inherit its attribution.
|
|
let bounded_lease_sql = deletion
|
|
.split_once("async fn bounded_serving_lease_sql<T>(")
|
|
.expect("deletion store must expose the bounded serving-lease helper")
|
|
.1
|
|
.split_once("/// Acquire a durable, expiring lease")
|
|
.expect("bounded serving-lease helper must precede lease acquisition")
|
|
.0;
|
|
assert!(bounded_lease_sql.contains("WriterOperation::EventWrite"));
|
|
assert!(!bounded_lease_sql.contains("self.pool.begin().await"));
|
|
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")
|
|
|| function.contains("self.bounded_serving_lease_sql("),
|
|
"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}"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn pool_level_insert_mentions_opens_the_tenant_local_chokepoint() {
|
|
let runtime = include_str!("../src/runtime/mod.rs");
|
|
let insert_mentions = runtime
|
|
.split_once("pub async fn insert_mentions(\n")
|
|
.expect("runtime must expose pool-level insert_mentions")
|
|
.1
|
|
.split_once("pub(crate) async fn insert_mentions_in_transaction(")
|
|
.expect("pool-level insert_mentions must precede its transaction seam")
|
|
.0;
|
|
assert!(
|
|
insert_mentions.contains(COMMUNITY_CHOKEPOINT_MARKER),
|
|
"post-commit mention indexing must be admitted like any other event-table write"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn event_write_paths_include_tenant_local_chokepoint_calls() {
|
|
let event = include_str!("../src/store/event.rs");
|
|
let insert_event = event
|
|
.split_once("pub async fn insert_event(\n")
|
|
.expect("event store must expose pool-level insert_event")
|
|
.1
|
|
.split_once("/// Insert a Nostr event in a caller-owned admitted transaction")
|
|
.expect("pool insert must precede transaction-seam insert")
|
|
.0;
|
|
assert!(
|
|
has_any_tenant_local_chokepoint(insert_event),
|
|
"pool-level event inserts must include a tenant-local event-write chokepoint call"
|
|
);
|
|
let insert_with_thread_meta = event
|
|
.split_once("pub async fn insert_event_with_thread_metadata(\n")
|
|
.expect("event store must expose pool-level thread-metadata insert")
|
|
.1
|
|
.split_once("impl Db {")
|
|
.expect("pool thread-metadata insert must precede Db wrappers")
|
|
.0;
|
|
assert!(
|
|
has_any_tenant_local_chokepoint(insert_with_thread_meta),
|
|
"thread-metadata event inserts must include a tenant-local chokepoint call"
|
|
);
|
|
|
|
let replaceable = include_str!("../src/store/replaceable.rs");
|
|
let replace_addressable = replaceable
|
|
.split_once("pub async fn replace_addressable_event(\n")
|
|
.expect("replaceable store must expose replace_addressable_event")
|
|
.1
|
|
.split_once("/// Replace a NIP-33 event inside a caller-owned transaction.")
|
|
.expect("addressable replacement must precede parameterized transaction seam")
|
|
.0;
|
|
assert!(
|
|
has_any_tenant_local_chokepoint(replace_addressable),
|
|
"addressable replacements must include a tenant-local chokepoint call"
|
|
);
|
|
let replace_parameterized = replaceable
|
|
.split_once("pub async fn replace_parameterized_event(\n")
|
|
.expect("replaceable store must expose replace_parameterized_event")
|
|
.1
|
|
.split_once("}\n\n#[cfg(test)]")
|
|
.expect("parameterized replacement must precede tests")
|
|
.0;
|
|
assert!(
|
|
has_any_tenant_local_chokepoint(replace_parameterized),
|
|
"parameterized replacements must include a tenant-local chokepoint call"
|
|
);
|
|
|
|
let channel_members = include_str!("../src/store/channel_members.rs");
|
|
let snapshot_lock = channel_members
|
|
.split_once("pub async fn lock_member_snapshot(\n")
|
|
.expect("channel_members must expose lock_member_snapshot")
|
|
.1
|
|
.split_once("/// Add a member to a channel.")
|
|
.expect("snapshot lock path must precede member add path")
|
|
.0;
|
|
assert!(
|
|
has_any_tenant_local_chokepoint(snapshot_lock),
|
|
"snapshot publication locks must include a tenant-local chokepoint call"
|
|
);
|
|
|
|
let push = include_str!("../src/store/push.rs");
|
|
let accept_lease = push
|
|
.split_once("pub async fn accept_lease_event(\n")
|
|
.expect("push store must expose accept_lease_event")
|
|
.1
|
|
.split_once("fn constraint_acceptance_outcome")
|
|
.expect("accept_lease_event must precede constraint outcome mapping")
|
|
.0;
|
|
assert!(
|
|
has_any_tenant_local_chokepoint(accept_lease),
|
|
"push lease source-event writes must include a tenant-local chokepoint call"
|
|
);
|
|
|
|
let reaction = include_str!("../src/store/reaction.rs");
|
|
let insert_reaction = reaction
|
|
.split_once("pub async fn insert_reaction_event_with_thread_metadata(\n")
|
|
.expect("reaction store must expose insert_reaction_event_with_thread_metadata")
|
|
.1
|
|
.split_once("/// Soft-delete a reaction by setting")
|
|
.expect("reaction insert must precede reaction soft-delete")
|
|
.0;
|
|
assert!(
|
|
has_any_tenant_local_chokepoint(insert_reaction),
|
|
"live kind:7 reaction inserts must include a tenant-local chokepoint call"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn legacy_compatibility_metrics_remain_pinned_to_the_preexisting_event_write_entrypoints() {
|
|
let runtime = include_str!("../src/runtime/mod.rs");
|
|
let typed_helper = runtime
|
|
.split_once("pub(crate) async fn begin_community_event_write_transaction(\n")
|
|
.expect("runtime must expose the typed tenant-local chokepoint")
|
|
.1
|
|
.split_once(
|
|
"pub(crate) async fn begin_community_event_write_transaction_with_legacy_metrics(\n",
|
|
)
|
|
.expect("typed chokepoint must precede the legacy compatibility wrapper")
|
|
.0;
|
|
assert!(
|
|
typed_helper.contains("CommunityEventWriteMetricPopulation::TypedOnly"),
|
|
"the default tenant-local chokepoint must stay typed-only"
|
|
);
|
|
assert!(
|
|
!typed_helper.contains("CommunityEventWriteMetricPopulation::LegacyCompatibility"),
|
|
"the default tenant-local chokepoint must not emit legacy compatibility metrics"
|
|
);
|
|
|
|
let legacy_db_wrapper = runtime
|
|
.split_once("pub async fn begin_event_write_transaction(\n")
|
|
.expect("Db must expose the public event-write entrypoint")
|
|
.1
|
|
.split_once("/// Insert an event while holding and validating an admitted serving-write")
|
|
.expect("public Db entrypoint must precede the serving-lease writer")
|
|
.0;
|
|
assert!(
|
|
legacy_db_wrapper.contains(COMMUNITY_CHOKEPOINT_LEGACY_MARKER),
|
|
"Db::begin_event_write_transaction must preserve the legacy compatibility population"
|
|
);
|
|
|
|
let replaceable = include_str!("../src/store/replaceable.rs");
|
|
for (label, start, end) in [
|
|
(
|
|
"replace_addressable_event",
|
|
"pub async fn replace_addressable_event(\n",
|
|
"/// Atomically replace a NIP-33 parameterized replaceable event.",
|
|
),
|
|
(
|
|
"replace_parameterized_event",
|
|
"pub async fn replace_parameterized_event(\n",
|
|
"}\n\n#[cfg(test)]",
|
|
),
|
|
] {
|
|
let seam = replaceable
|
|
.split_once(start)
|
|
.unwrap_or_else(|| panic!("replaceable store must expose {label}"))
|
|
.1
|
|
.split_once(end)
|
|
.unwrap_or_else(|| panic!("{label} must precede its next production seam"))
|
|
.0;
|
|
assert!(
|
|
seam.contains(COMMUNITY_CHOKEPOINT_LEGACY_MARKER),
|
|
"{label} must preserve the legacy compatibility population"
|
|
);
|
|
}
|
|
|
|
let push = include_str!("../src/store/push.rs");
|
|
let accept_lease = push
|
|
.split_once("pub async fn accept_lease_event(\n")
|
|
.expect("push store must expose accept_lease_event")
|
|
.1
|
|
.split_once("fn constraint_acceptance_outcome")
|
|
.expect("accept_lease_event must precede constraint outcome mapping")
|
|
.0;
|
|
assert!(
|
|
accept_lease.contains(COMMUNITY_CHOKEPOINT_LEGACY_MARKER),
|
|
"push lease acceptance must preserve the legacy compatibility population"
|
|
);
|
|
|
|
let event = include_str!("../src/store/event.rs");
|
|
let insert_event = event
|
|
.split_once("pub async fn insert_event(\n")
|
|
.expect("event store must expose pool-level insert_event")
|
|
.1
|
|
.split_once("/// Insert a Nostr event in a caller-owned admitted transaction")
|
|
.expect("pool insert must precede transaction-seam insert")
|
|
.0;
|
|
assert!(
|
|
insert_event.contains(COMMUNITY_CHOKEPOINT_MARKER),
|
|
"typed-only event inserts must stay on the typed tenant-local chokepoint"
|
|
);
|
|
assert!(
|
|
!insert_event.contains(COMMUNITY_CHOKEPOINT_LEGACY_MARKER),
|
|
"typed-only event inserts must not emit legacy compatibility metrics"
|
|
);
|
|
|
|
let reaction = include_str!("../src/store/reaction.rs");
|
|
let insert_reaction = reaction
|
|
.split_once("pub async fn insert_reaction_event_with_thread_metadata(\n")
|
|
.expect("reaction store must expose insert_reaction_event_with_thread_metadata")
|
|
.1
|
|
.split_once("/// Soft-delete a reaction by setting")
|
|
.expect("reaction insert must precede reaction soft-delete")
|
|
.0;
|
|
assert!(
|
|
insert_reaction.contains(COMMUNITY_CHOKEPOINT_MARKER),
|
|
"typed-only reaction inserts must stay on the typed tenant-local chokepoint"
|
|
);
|
|
assert!(
|
|
!insert_reaction.contains(COMMUNITY_CHOKEPOINT_LEGACY_MARKER),
|
|
"typed-only reaction inserts must not emit legacy compatibility metrics"
|
|
);
|
|
}
|
|
|
|
/// Function-level routing check for guarded-table writers.
|
|
///
|
|
/// A file can contain both a legitimate chokepoint writer and a bypass writer.
|
|
/// Every writing function must either open the tenant-local community
|
|
/// chokepoint itself, take a caller-owned `&mut AdmittedTx` (which only the
|
|
/// chokepoint can construct, so the compiler proves provenance), use reviewed
|
|
/// adapter-owned transaction state whose constructor is pinned to the
|
|
/// chokepoint, or appear in a reviewed exception list below.
|
|
///
|
|
/// A raw `&mut Transaction` or `&mut PgConnection` parameter is not a route:
|
|
/// nothing about its type says the transaction was admitted.
|
|
const GUARDED_TABLE_WRITE_MARKERS: [&str; 9] = [
|
|
"INSERT INTO events",
|
|
"UPDATE events",
|
|
"DELETE FROM events",
|
|
"INSERT INTO reactions",
|
|
"UPDATE reactions",
|
|
"DELETE FROM reactions",
|
|
"INSERT INTO event_mentions",
|
|
"UPDATE event_mentions",
|
|
"DELETE FROM event_mentions",
|
|
];
|
|
|
|
const COMMUNITY_CHOKEPOINT_MARKER: &str = "begin_community_event_write_transaction(";
|
|
const GENERAL_COMMUNITY_CHOKEPOINT_MARKER: &str = "begin_community_write_transaction(";
|
|
const SWEEP_COMMUNITY_CHOKEPOINT_MARKER: &str = "begin_community_sweep_transaction(";
|
|
const COMMUNITY_CHOKEPOINT_LEGACY_MARKER: &str =
|
|
"begin_community_event_write_transaction_with_legacy_metrics(";
|
|
|
|
fn has_any_tenant_local_chokepoint(source: &str) -> bool {
|
|
source.contains(COMMUNITY_CHOKEPOINT_MARKER)
|
|
|| source.contains(SWEEP_COMMUNITY_CHOKEPOINT_MARKER)
|
|
|| source.contains(GENERAL_COMMUNITY_CHOKEPOINT_MARKER)
|
|
|| source.contains(COMMUNITY_CHOKEPOINT_LEGACY_MARKER)
|
|
}
|
|
|
|
const ADMITTED_TX_SIGNATURE_MARKERS: [&str; 2] = ["&mut AdmittedTx", "&mut crate::AdmittedTx"];
|
|
|
|
// Narrow reviewed exceptions for non-serving verification probes only.
|
|
const GUARDED_WRITE_FUNCTION_EXCEPTIONS: [&str; 1] = ["pub async fn verify_floor_guard_behavior("];
|
|
|
|
const GUARDED_TX_ADAPTER_METHOD_PINS: [(&str, &str); 1] = [(
|
|
"pub async fn replace_member_event(",
|
|
"pub async fn lock_member_snapshot(",
|
|
)];
|
|
|
|
fn production_contains_guarded_write(production_source: &str) -> bool {
|
|
GUARDED_TABLE_WRITE_MARKERS
|
|
.iter()
|
|
.any(|marker| production_source.contains(marker))
|
|
}
|
|
|
|
fn fn_signature_starts_here(trimmed_line: &str) -> bool {
|
|
[
|
|
"fn ",
|
|
"async fn ",
|
|
"pub fn ",
|
|
"pub async fn ",
|
|
"pub(crate) fn ",
|
|
"pub(crate) async fn ",
|
|
"pub(super) fn ",
|
|
"pub(super) async fn ",
|
|
]
|
|
.iter()
|
|
.any(|prefix| trimmed_line.starts_with(prefix))
|
|
}
|
|
|
|
/// Split production source into one slice per function. A slice runs from its
|
|
/// signature to the doc comment or attributes of the next function, so a
|
|
/// following function's docs (which may quote SQL) are never attributed to the
|
|
/// function before it.
|
|
///
|
|
/// Signatures and attributes are found in the masked view (see
|
|
/// [`mask_comments_and_literals`]), so an `fn` line inside a comment or a
|
|
/// multiline string never splits a function. Doc comments are blank in that
|
|
/// view, so `///` lead-ins are read from the original line.
|
|
fn function_slices(production_source: &str) -> Vec<&str> {
|
|
let masked = mask_comments_and_literals(production_source);
|
|
// (signature offset, offset where the next function's lead-in begins)
|
|
let mut starts = Vec::new();
|
|
let mut lead_ins = Vec::new();
|
|
let mut lead_in: Option<usize> = None;
|
|
let mut offset = 0usize;
|
|
for (line, code) in production_source
|
|
.split_inclusive('\n')
|
|
.zip(masked.split_inclusive('\n'))
|
|
{
|
|
let trimmed = code.trim_start();
|
|
let indent = code.len() - trimmed.len();
|
|
if fn_signature_starts_here(trimmed) {
|
|
starts.push(offset + indent);
|
|
lead_ins.push(lead_in.take().unwrap_or(offset + indent));
|
|
} else if line.trim_start().starts_with("///") || trimmed.starts_with("#[") {
|
|
lead_in.get_or_insert(offset);
|
|
} else {
|
|
lead_in = None;
|
|
}
|
|
offset += line.len();
|
|
}
|
|
|
|
let mut functions = Vec::new();
|
|
for (index, start) in starts.iter().enumerate() {
|
|
let end = lead_ins
|
|
.get(index + 1)
|
|
.copied()
|
|
.unwrap_or(production_source.len());
|
|
functions.push(&production_source[*start..end]);
|
|
}
|
|
functions
|
|
}
|
|
|
|
fn function_header(function_source: &str) -> &str {
|
|
function_source
|
|
.lines()
|
|
.next()
|
|
.unwrap_or("<unknown function>")
|
|
.trim()
|
|
}
|
|
|
|
fn function_is_guarded_write_exception(function_header: &str) -> bool {
|
|
GUARDED_WRITE_FUNCTION_EXCEPTIONS
|
|
.iter()
|
|
.any(|exception| function_header.starts_with(exception))
|
|
}
|
|
|
|
fn function_accepts_admitted_transaction(function_source: &str) -> bool {
|
|
let signature = function_source.split('{').next().unwrap_or(function_source);
|
|
ADMITTED_TX_SIGNATURE_MARKERS
|
|
.iter()
|
|
.any(|marker| signature.contains(marker))
|
|
}
|
|
|
|
fn function_uses_guarded_tx_adapter_state(function_source: &str) -> bool {
|
|
function_source.contains("&mut *self.tx") || function_source.contains("&mut self.tx")
|
|
}
|
|
|
|
fn adapter_constructor_is_chokepoint_pinned(production_source: &str, method_header: &str) -> bool {
|
|
for (adapter_method, adapter_constructor) in GUARDED_TX_ADAPTER_METHOD_PINS {
|
|
if method_header.starts_with(adapter_method) {
|
|
return function_slices(production_source)
|
|
.into_iter()
|
|
.find(|function_source| {
|
|
function_header(function_source).starts_with(adapter_constructor)
|
|
})
|
|
.is_some_and(has_any_tenant_local_chokepoint);
|
|
}
|
|
}
|
|
|
|
false
|
|
}
|
|
|
|
fn function_has_syntactic_guarded_write_route(
|
|
function_source: &str,
|
|
production_source: &str,
|
|
) -> bool {
|
|
let header = function_header(function_source);
|
|
|
|
has_any_tenant_local_chokepoint(function_source)
|
|
|| function_accepts_admitted_transaction(function_source)
|
|
|| (function_uses_guarded_tx_adapter_state(function_source)
|
|
&& adapter_constructor_is_chokepoint_pinned(production_source, header))
|
|
}
|
|
|
|
fn syntactic_guarded_write_route_violations(production_source: &str) -> Vec<String> {
|
|
function_slices(production_source)
|
|
.into_iter()
|
|
.filter(|function_source| {
|
|
let header = function_header(function_source);
|
|
production_contains_guarded_write(function_source)
|
|
&& !function_is_guarded_write_exception(header)
|
|
&& !function_has_syntactic_guarded_write_route(function_source, production_source)
|
|
})
|
|
.map(|function_source| function_header(function_source).to_owned())
|
|
.collect()
|
|
}
|
|
|
|
#[test]
|
|
fn serving_table_policy_rejects_same_file_mixed_guarded_writers() {
|
|
let mixed_source = r#"
|
|
pub async fn guarded_writer(pool: &sqlx::PgPool) {
|
|
let mut tx = begin_community_event_write_transaction(pool, community, WriterOperation::EventWrite)
|
|
.await
|
|
.expect("tx");
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(&mut *tx)
|
|
.await
|
|
.expect("write");
|
|
}
|
|
pub async fn unguarded_writer(pool: &sqlx::PgPool) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(pool)
|
|
.await
|
|
.expect("unguarded");
|
|
}
|
|
"#;
|
|
|
|
let violations = syntactic_guarded_write_route_violations(mixed_source);
|
|
assert!(
|
|
violations
|
|
.iter()
|
|
.any(|name| name.starts_with("pub async fn unguarded_writer(")),
|
|
"whole-file co-occurrence policy is too weak: one guarded writer in a file must not bless \
|
|
a sibling unguarded writer; violations: {violations:?}"
|
|
);
|
|
assert!(
|
|
!violations
|
|
.iter()
|
|
.any(|name| name.starts_with("pub async fn guarded_writer(")),
|
|
"guarded writer must remain allowed; violations: {violations:?}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn serving_table_policy_rejects_writer_connection_only_bypasses() {
|
|
let acquire_writer_bypass = r#"
|
|
pub async fn bypass_with_writer_connection(pool: &sqlx::PgPool) {
|
|
let mut connection = crate::observability::acquire_writer(pool, WriterOperation::EventWrite)
|
|
.await
|
|
.expect("connection");
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(&mut *connection)
|
|
.await
|
|
.expect("write");
|
|
}
|
|
"#;
|
|
|
|
let pool_begin_bypass = r#"
|
|
pub async fn bypass_with_pool_begin(pool: &sqlx::PgPool) {
|
|
let mut tx = pool.begin().await.expect("tx");
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(&mut *tx)
|
|
.await
|
|
.expect("write");
|
|
}
|
|
"#;
|
|
|
|
let acquire_writer_violations = syntactic_guarded_write_route_violations(acquire_writer_bypass);
|
|
assert!(
|
|
acquire_writer_violations
|
|
.iter()
|
|
.any(|name| name.starts_with("pub async fn bypass_with_writer_connection(")),
|
|
"acquire_writer + raw guarded-table write must be rejected; violations: \
|
|
{acquire_writer_violations:?}"
|
|
);
|
|
|
|
let pool_begin_violations = syntactic_guarded_write_route_violations(pool_begin_bypass);
|
|
assert!(
|
|
pool_begin_violations
|
|
.iter()
|
|
.any(|name| name.starts_with("pub async fn bypass_with_pool_begin(")),
|
|
"pool.begin + raw guarded-table write must be rejected; violations: \
|
|
{pool_begin_violations:?}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn serving_table_policy_requires_admitted_tx_not_raw_transactions() {
|
|
let source = r#"
|
|
pub(crate) async fn raw_transaction_writer(tx: &mut sqlx::Transaction<'_, sqlx::Postgres>) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
}
|
|
pub(crate) async fn raw_connection_writer(conn: &mut sqlx::PgConnection) {
|
|
sqlx::query("INSERT INTO event_mentions (community_id, event_id) VALUES ($1, $2)")
|
|
.execute(conn)
|
|
.await
|
|
.expect("write");
|
|
}
|
|
pub(crate) async fn admitted_writer(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.bind(tx.community().as_uuid())
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
}
|
|
"#;
|
|
|
|
let violations = syntactic_guarded_write_route_violations(source);
|
|
assert_eq!(
|
|
violations,
|
|
[
|
|
"pub(crate) async fn raw_transaction_writer(tx: &mut sqlx::Transaction<'_, sqlx::Postgres>) {",
|
|
"pub(crate) async fn raw_connection_writer(conn: &mut sqlx::PgConnection) {",
|
|
],
|
|
"a raw transaction or connection parameter carries no admission proof; only \
|
|
`&mut AdmittedTx` does"
|
|
);
|
|
}
|
|
|
|
/// Every function in `source` that builds an `AdmittedTx`, paired with whether
|
|
/// it admits the transaction it wraps. The scan deliberately does not strip
|
|
/// `#[cfg(test)]` items: a test-only forge would let crate tests run follow-ups
|
|
/// on an unadmitted transaction, so it must admit like any other constructor.
|
|
fn admitted_tx_constructors(source: &str) -> Vec<(String, bool)> {
|
|
let production: String = source
|
|
.lines()
|
|
.filter(|line| !line.trim_start().starts_with("//"))
|
|
// `impl … for AdmittedTx {` headers open a block, not a value.
|
|
.filter(|line| !line.trim_start().starts_with("impl"))
|
|
.map(|line| format!("{line}\n"))
|
|
.collect();
|
|
function_slices(&production)
|
|
.into_iter()
|
|
.filter(|function| function.contains("Self {") || function.contains("AdmittedTx {"))
|
|
.map(|function| {
|
|
let admits = function.contains(".guard_transaction(&mut tx, community)")
|
|
|| function.contains(".guard_transaction_with_serving_lease(&mut tx, lease)");
|
|
(function.to_string(), admits)
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
#[test]
|
|
fn admitted_tx_is_constructed_only_by_admitting_constructors() {
|
|
// The fields are private to `runtime/admitted_tx.rs`, so only that file can
|
|
// build the value. Pin that each function there that builds it also admits
|
|
// the transaction it wraps, so a new unguarded constructor cannot slip in.
|
|
let constructors = admitted_tx_constructors(include_str!("../src/runtime/admitted_tx.rs"));
|
|
assert_eq!(
|
|
constructors.len(),
|
|
2,
|
|
"AdmittedTx must have exactly the two admitting constructors"
|
|
);
|
|
for (constructor, admits) in constructors {
|
|
assert!(
|
|
admits,
|
|
"AdmittedTx constructor must admit the transaction it wraps: {constructor}"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn admitted_tx_constructor_scan_reports_test_only_forges() {
|
|
let source = r#"
|
|
impl AdmittedTx {
|
|
pub(crate) async fn admit(mut tx: Tx, community: CommunityId) -> Result<Self> {
|
|
guard.guard_transaction(&mut tx, community).await?;
|
|
Ok(Self { tx, community })
|
|
}
|
|
|
|
#[cfg(test)]
|
|
pub(crate) fn forge_for_tests(tx: Tx, community: CommunityId) -> Self {
|
|
Self { tx, community }
|
|
}
|
|
}
|
|
"#;
|
|
let unadmitted: Vec<String> = admitted_tx_constructors(source)
|
|
.into_iter()
|
|
.filter(|(_, admits)| !admits)
|
|
.map(|(function, _)| {
|
|
function
|
|
.lines()
|
|
.next()
|
|
.unwrap_or_default()
|
|
.trim()
|
|
.to_string()
|
|
})
|
|
.collect();
|
|
assert_eq!(
|
|
unadmitted,
|
|
["pub(crate) fn forge_for_tests(tx: Tx, community: CommunityId) -> Self {"],
|
|
"an indented `#[cfg(test)]` forge must still be reported as unadmitted"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn cfg_test_items_are_skipped_without_hiding_later_production_code() {
|
|
let source = "pub fn before() {}\n\
|
|
#[cfg(test)]\n\
|
|
struct Marker;\n\
|
|
pub fn after_struct() {}\n\
|
|
#[cfg(test)]\n\
|
|
static LOCK: std::sync::Mutex<()> =\n\
|
|
std::sync::Mutex::new(());\n\
|
|
pub fn after_static() {}\n\
|
|
#[cfg(test)]\n\
|
|
static S: [u8; 1] =\n\
|
|
[const { 0 }; 1];\n\
|
|
pub fn after_const_block() {}\n\
|
|
#[cfg(test)]\n\
|
|
#[derive(Debug)]\n\
|
|
struct Fields {\n\
|
|
value: u8,\n\
|
|
}\n\
|
|
pub fn after_fields() {}\n\
|
|
#[cfg(test)]\n\
|
|
mod tests {\n\
|
|
fn hidden() {\n\
|
|
}\n\
|
|
}\n\
|
|
pub fn after_module() {}\n";
|
|
let production = strip_cfg_test_items(source);
|
|
for name in [
|
|
"before",
|
|
"after_struct",
|
|
"after_static",
|
|
"after_const_block",
|
|
"after_fields",
|
|
"after_module",
|
|
] {
|
|
assert!(
|
|
production.contains(&format!("pub fn {name}()")),
|
|
"{name} is production code and must stay visible: {production}"
|
|
);
|
|
}
|
|
for hidden in ["Marker", "LOCK", "static S", "value: u8", "fn hidden"] {
|
|
assert!(
|
|
!production.contains(hidden),
|
|
"{hidden} is test-only and must be skipped: {production}"
|
|
);
|
|
}
|
|
|
|
let raw_writer = "pub(crate) async fn raw_writer(conn: &mut sqlx::PgConnection) {\n\
|
|
sqlx::query(\"INSERT INTO events (community_id, id) VALUES ($1, $2)\")\n\
|
|
.execute(conn)\n\
|
|
.await\n\
|
|
.expect(\"write\");\n\
|
|
}\n";
|
|
for (test_item, hidden) in [
|
|
("struct X;", "struct X"),
|
|
("fn helper() {} // test helper", "fn helper"),
|
|
("const BRACES: &str = \"{}\"; // fixture", "BRACES"),
|
|
(
|
|
"const URL: &str = \"https://relay.test\"; // fixture",
|
|
"relay.test",
|
|
),
|
|
("fn u() -> &'static str { \"ws://x\" } // c", "ws://x"),
|
|
] {
|
|
let production = strip_cfg_test_items(&format!("#[cfg(test)]\n{test_item}\n{raw_writer}"));
|
|
assert!(
|
|
!production.contains(hidden),
|
|
"`{test_item}` is test-only and must be skipped: {production}"
|
|
);
|
|
assert!(
|
|
production_contains_guarded_write(&production),
|
|
"a guarded write after `{test_item}` must stay visible: {production}"
|
|
);
|
|
assert_eq!(
|
|
syntactic_guarded_write_route_violations(&production),
|
|
["pub(crate) async fn raw_writer(conn: &mut sqlx::PgConnection) {"],
|
|
"a raw writer after `{test_item}` must still be scanned"
|
|
);
|
|
}
|
|
}
|
|
|
|
fn should_scan_guarded_write_source_file(path: &std::path::Path) -> bool {
|
|
let file_name = path
|
|
.file_name()
|
|
.and_then(|name| name.to_str())
|
|
.unwrap_or("");
|
|
path.extension().is_some_and(|ext| ext == "rs")
|
|
// Whole files loaded only via `#[cfg(test)] #[path = "..."] mod ...;` are
|
|
// entirely test code but carry no internal `#[cfg(test)]` marker of their
|
|
// own to slice against. Standalone `*_tests.rs` modules share that shape.
|
|
&& file_name != "tests.rs"
|
|
&& !file_name.ends_with("_tests.rs")
|
|
}
|
|
|
|
#[test]
|
|
fn serving_table_policy_skips_standalone_test_modules() {
|
|
use std::path::Path;
|
|
|
|
assert!(
|
|
!should_scan_guarded_write_source_file(Path::new("src/runtime/tests.rs")),
|
|
"`tests.rs` is a standalone test-only module"
|
|
);
|
|
assert!(
|
|
!should_scan_guarded_write_source_file(Path::new(
|
|
"src/store/thread_window/postgres_tests.rs",
|
|
)),
|
|
"`*_tests.rs` modules are standalone test-only sources, not production seams"
|
|
);
|
|
assert!(
|
|
!should_scan_guarded_write_source_file(Path::new("src/store/foo_tests.rs")),
|
|
"the suffix-based rule must cover other standalone test-only modules"
|
|
);
|
|
assert!(
|
|
should_scan_guarded_write_source_file(Path::new("src/store/event.rs")),
|
|
"production source must remain in scope"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn serving_table_writes_expose_syntactic_chokepoint_or_guarded_tx_routes() {
|
|
use std::path::{Path, PathBuf};
|
|
|
|
fn collect_rs_files(dir: &Path, out: &mut Vec<PathBuf>) {
|
|
for entry in std::fs::read_dir(dir).expect("read source directory") {
|
|
let entry = entry.expect("read directory entry");
|
|
let path = entry.path();
|
|
if path.is_dir() {
|
|
collect_rs_files(&path, out);
|
|
} else if should_scan_guarded_write_source_file(&path) {
|
|
out.push(path);
|
|
}
|
|
}
|
|
}
|
|
|
|
// `buzz-relay` is scanned too: relay code reaches fenced tables only through
|
|
// `buzz-db` helpers, and a direct relay write must meet the same rule.
|
|
let manifest = Path::new(env!("CARGO_MANIFEST_DIR"));
|
|
let mut checked_guarded_files = 0usize;
|
|
for (crate_name, src_root) in [
|
|
("buzz-db", manifest.join("src")),
|
|
("buzz-relay", manifest.join("../buzz-relay/src")),
|
|
] {
|
|
let mut files = Vec::new();
|
|
collect_rs_files(&src_root, &mut files);
|
|
assert!(
|
|
!files.is_empty(),
|
|
"guarded-table scan must see {crate_name} production source files"
|
|
);
|
|
for path in files {
|
|
let relative = path
|
|
.strip_prefix(&src_root)
|
|
.expect("file is under src root")
|
|
.to_string_lossy()
|
|
.replace('\\', "/");
|
|
let source = std::fs::read_to_string(&path).expect("read source file");
|
|
let production = strip_cfg_test_items(&source);
|
|
if production_contains_guarded_write(&production) {
|
|
checked_guarded_files += 1;
|
|
let violations = syntactic_guarded_write_route_violations(&production);
|
|
assert!(
|
|
violations.is_empty(),
|
|
"{crate_name}/src/{relative} has guarded-table INSERT/UPDATE/DELETE seams \
|
|
that neither open the tenant-local chokepoint nor take `&mut AdmittedTx`: \
|
|
{violations:?}"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
assert!(
|
|
checked_guarded_files > 0,
|
|
"guarded-table scan must exercise at least one file that writes a fenced table"
|
|
);
|
|
}
|
|
|
|
// Rolled-back verification probes write `events` rows that never commit, so
|
|
// they owe no push job or TTL refresh.
|
|
const EVENT_INSERT_FOLLOW_UP_EXCEPTIONS: [&str; 1] = ["pub async fn verify_floor_guard_behavior("];
|
|
|
|
/// `source` with every `#[cfg(test)]` item removed, wherever it sits in the
|
|
/// file. Each item runs to the `;` that ends it or to the brace that closes
|
|
/// its body. Braces inside comments and literals are not counted (see
|
|
/// [`mask_comments_and_literals`]). Unlike cutting the file at the first
|
|
/// `#[cfg(test)]`, this keeps production code that follows a test-only item.
|
|
/// The whole file is masked once and the marker is found in that view, so a
|
|
/// mention in a comment or string (even a multiline one) strips nothing. The
|
|
/// marker also counts only when it is the first code on its line.
|
|
fn strip_cfg_test_items(source: &str) -> String {
|
|
const MARKER: &str = "#[cfg(test)]";
|
|
let masked = mask_comments_and_literals(source);
|
|
let mut kept = String::with_capacity(source.len());
|
|
let mut copied = 0usize;
|
|
let mut search = 0usize;
|
|
while let Some(found) = masked[search..].find(MARKER) {
|
|
let at = search + found;
|
|
let line_start = masked[..at].rfind('\n').map_or(0, |newline| newline + 1);
|
|
search = at + MARKER.len();
|
|
if !masked[line_start..at].trim().is_empty() {
|
|
continue;
|
|
}
|
|
kept.push_str(&source[copied..at]);
|
|
copied = search + cfg_test_item_len(&masked[search..]);
|
|
search = copied;
|
|
}
|
|
kept.push_str(&source[copied..]);
|
|
kept
|
|
}
|
|
|
|
/// Length of the item that starts `masked_item`, which must already be masked
|
|
/// so braces in comments and literals are not counted.
|
|
fn cfg_test_item_len(masked_item: &str) -> usize {
|
|
let mut depth = 0usize;
|
|
for (index, byte) in masked_item.bytes().enumerate() {
|
|
match byte {
|
|
b';' if depth == 0 => return index + 1,
|
|
b'{' => depth += 1,
|
|
b'}' => {
|
|
depth = depth.saturating_sub(1);
|
|
if depth == 0 {
|
|
return index + 1;
|
|
}
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
masked_item.len()
|
|
}
|
|
|
|
/// Whether the `r` at `index` begins a raw literal token: `r"`, or the `br"` /
|
|
/// `cr"` raw byte and C strings, rather than ending an identifier such as `for"`.
|
|
fn raw_literal_prefix_starts_token(bytes: &[u8], index: usize) -> bool {
|
|
let is_ident = |at: Option<usize>| {
|
|
at.and_then(|at| bytes.get(at))
|
|
.is_some_and(|byte| byte.is_ascii_alphanumeric() || *byte == b'_')
|
|
};
|
|
let before = index.checked_sub(1);
|
|
!is_ident(before)
|
|
|| (matches!(before.map(|at| bytes[at]), Some(b'b' | b'c'))
|
|
&& !is_ident(index.checked_sub(2)))
|
|
}
|
|
|
|
/// `source` with every comment (`//`, `///`, nested `/* */`) and every string,
|
|
/// raw string, and char literal replaced by spaces, byte for byte. Newlines are
|
|
/// kept, so offsets and line structure match `source`. Scans that look for
|
|
/// code (braces, calls) read the masked text; scans that look for SQL read
|
|
/// the original, because the SQL lives inside string literals.
|
|
fn mask_comments_and_literals(source: &str) -> String {
|
|
let bytes = source.as_bytes();
|
|
let mut masked = bytes.to_vec();
|
|
let mut blank = |from: usize, to: usize| {
|
|
for byte in &mut masked[from..to.min(bytes.len())] {
|
|
if *byte != b'\n' {
|
|
*byte = b' ';
|
|
}
|
|
}
|
|
};
|
|
let mut index = 0usize;
|
|
while index < bytes.len() {
|
|
let start = index;
|
|
match bytes[index] {
|
|
b'/' if bytes.get(index + 1) == Some(&b'/') => {
|
|
while index < bytes.len() && bytes[index] != b'\n' {
|
|
index += 1;
|
|
}
|
|
}
|
|
b'/' if bytes.get(index + 1) == Some(&b'*') => {
|
|
let mut depth = 0usize;
|
|
while index < bytes.len() {
|
|
if bytes[index] == b'/' && bytes.get(index + 1) == Some(&b'*') {
|
|
depth += 1;
|
|
index += 2;
|
|
} else if bytes[index] == b'*' && bytes.get(index + 1) == Some(&b'/') {
|
|
depth -= 1;
|
|
index += 2;
|
|
if depth == 0 {
|
|
break;
|
|
}
|
|
} else {
|
|
index += 1;
|
|
}
|
|
}
|
|
}
|
|
b'r' if matches!(bytes.get(index + 1), Some(b'"' | b'#'))
|
|
&& raw_literal_prefix_starts_token(bytes, index)
|
|
&& bytes.get(
|
|
index
|
|
+ 1
|
|
+ bytes[index + 1..]
|
|
.iter()
|
|
.take_while(|byte| **byte == b'#')
|
|
.count(),
|
|
) == Some(&b'"') =>
|
|
{
|
|
let hashes = bytes[index + 1..]
|
|
.iter()
|
|
.take_while(|byte| **byte == b'#')
|
|
.count();
|
|
let close = format!("\"{}", "#".repeat(hashes));
|
|
let body = index + 2 + hashes;
|
|
index = source[body..]
|
|
.find(&close)
|
|
.map_or(bytes.len(), |offset| body + offset + close.len());
|
|
}
|
|
b'"' => {
|
|
index += 1;
|
|
while index < bytes.len() && bytes[index] != b'"' {
|
|
index += if bytes[index] == b'\\' { 2 } else { 1 };
|
|
}
|
|
index += 1;
|
|
}
|
|
b'\'' if char_literal_end(source, index).is_some() => {
|
|
index = char_literal_end(source, index).expect("guard matched");
|
|
}
|
|
_ => {
|
|
index += 1;
|
|
continue;
|
|
}
|
|
}
|
|
blank(start, index);
|
|
}
|
|
String::from_utf8(masked).expect("masking replaces whole ASCII-delimited spans")
|
|
}
|
|
|
|
/// The end (exclusive) of the char literal whose opening quote is at `start`,
|
|
/// such as `'{'`, `'é'`, `'\''`, `'\x7f'`, or `'\u{1F600}'`. `None` for a
|
|
/// lifetime, which has no closing quote.
|
|
fn char_literal_end(source: &str, start: usize) -> Option<usize> {
|
|
// `start` holds an ASCII quote, so `start + 1` is a char boundary.
|
|
let rest = &source[start + 1..];
|
|
let body = if let Some(escape) = rest.strip_prefix('\\') {
|
|
// The longest escape is `\u{10FFFF}`: a backslash and 9 more bytes.
|
|
// Skip the escaped byte so `'\''` does not close on its own quote.
|
|
let close = escape.get(1..)?.find('\'')? + 1;
|
|
(close <= 9).then_some(1 + close)?
|
|
} else {
|
|
rest.chars().next()?.len_utf8()
|
|
};
|
|
rest[body..]
|
|
.starts_with('\'')
|
|
.then_some(start + 1 + body + 1)
|
|
}
|
|
|
|
#[test]
|
|
fn masker_blanks_every_literal_form_and_keeps_code() {
|
|
for (input, expected) in [
|
|
// Raw strings with zero, one, or several `#`, and the `b`/`c` raw
|
|
// prefixes. Each body ends in `\`, which would escape the closing quote
|
|
// if the literal were scanned as a plain string.
|
|
(r#"a(r"\") {}"#, "a( ) {}"),
|
|
(r###"a(r##"x"#{"##) {}"###, "a( ) {}"),
|
|
(r##"a(br#"\"#) {}"##, "a(b ) {}"),
|
|
(r#"a(br"\") {}"#, "a(b ) {}"),
|
|
(r#"a(cr"\") {}"#, "a(c ) {}"),
|
|
// Plain byte and C strings.
|
|
(r#"a(b"{") {}"#, "a(b ) {}"),
|
|
(r#"a(c"{") {}"#, "a(c ) {}"),
|
|
// Plain and escaped char literals.
|
|
(
|
|
r"a('\'', '\\', '\n', '{') {}",
|
|
"a( , , , ) {}",
|
|
),
|
|
// Multi-byte and long-escape char literals. Each holds a brace, or
|
|
// hides one from a scan that misreads the literal's length.
|
|
("a('é', '{') {}", "a( , ) {}"),
|
|
(r"a('\u{1F600}', '{') {}", "a( , ) {}"),
|
|
(r"a('\x7f', '{') {}", "a( , ) {}"),
|
|
// Code that only looks like a literal stays as it is.
|
|
("let r#type = 1; {}", "let r#type = 1; {}"),
|
|
("fn f<'a>(x: &'a str) {}", "fn f<'a>(x: &'a str) {}"),
|
|
] {
|
|
assert_eq!(
|
|
mask_comments_and_literals(input),
|
|
expected,
|
|
"masking `{input}`"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// Whether `source` inserts into `events` itself (not `event_mentions` or any
|
|
/// other `events_*` table), in any letter case, across line breaks and string
|
|
/// continuations, `\n`/`\t`/`\r` escapes, and with or without a `public.`
|
|
/// schema prefix.
|
|
fn inserts_events_row(source: &str) -> bool {
|
|
!events_insert_tails(source).is_empty()
|
|
}
|
|
|
|
/// `source` lowercased, with every run of whitespace, `\` line continuations,
|
|
/// and `\n`/`\t`/`\r` escapes collapsed to one space.
|
|
fn normalized_sql(source: &str) -> String {
|
|
let mut normalized = String::with_capacity(source.len());
|
|
let mut chars = source.chars().peekable();
|
|
while let Some(ch) = chars.next() {
|
|
if ch.is_whitespace() || ch == '\\' {
|
|
if ch == '\\' && matches!(chars.peek(), Some('n' | 't' | 'r')) {
|
|
chars.next();
|
|
}
|
|
if !normalized.ends_with(' ') {
|
|
normalized.push(' ');
|
|
}
|
|
} else {
|
|
normalized.push(ch.to_ascii_lowercase());
|
|
}
|
|
}
|
|
normalized
|
|
}
|
|
|
|
/// The normalized SQL that follows each `INSERT INTO events` in `source`.
|
|
fn events_insert_tails(source: &str) -> Vec<String> {
|
|
let normalized = normalized_sql(source);
|
|
let mut tails = Vec::new();
|
|
for marker in ["insert into events", "insert into public.events"] {
|
|
for (index, marker) in normalized.match_indices(marker) {
|
|
let tail = &normalized[index + marker.len()..];
|
|
if tail
|
|
.chars()
|
|
.next()
|
|
.is_none_or(|next| !(next.is_ascii_alphanumeric() || next == '_'))
|
|
{
|
|
tails.push(tail.to_owned());
|
|
}
|
|
}
|
|
}
|
|
tails
|
|
}
|
|
|
|
/// An events insert must run both follow-ups: the push enqueue and the
|
|
/// pre-commit TTL refresh. `after_admitted_insert` does both; a writer that
|
|
/// inserts inside a savepoint calls the two halves separately.
|
|
///
|
|
/// Only real calls count: a hook in a comment or a string literal does not.
|
|
fn runs_event_follow_ups(function_source: &str) -> bool {
|
|
let code = mask_comments_and_literals(function_source);
|
|
code.contains("event_follow_up::after_admitted_insert(")
|
|
|| (code.contains("event_follow_up::enqueue_push_match(")
|
|
&& code.contains(".record_channel_event("))
|
|
}
|
|
|
|
fn event_insert_follow_up_violations(production_source: &str) -> Vec<String> {
|
|
function_slices(production_source)
|
|
.into_iter()
|
|
.filter(|function_source| {
|
|
let header = function_header(function_source);
|
|
inserts_events_row(function_source)
|
|
&& !EVENT_INSERT_FOLLOW_UP_EXCEPTIONS
|
|
.iter()
|
|
.any(|exception| header.starts_with(exception))
|
|
&& !runs_event_follow_ups(function_source)
|
|
})
|
|
.map(|function_source| function_header(function_source).to_owned())
|
|
.collect()
|
|
}
|
|
|
|
#[test]
|
|
fn event_insert_follow_up_policy_rejects_writers_without_the_hook() {
|
|
let source = r##"
|
|
pub(crate) async fn unhooked_writer(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
}
|
|
pub(crate) async fn push_only_writer(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events \
|
|
(community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
crate::store::event_follow_up::enqueue_push_match(tx.conn(), community, id, kind)
|
|
.await
|
|
.expect("enqueue");
|
|
}
|
|
pub(crate) async fn hooked_writer(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
crate::store::event_follow_up::after_admitted_insert(tx, id, kind, channel)
|
|
.await
|
|
.expect("follow up");
|
|
}
|
|
pub(crate) async fn savepoint_writer(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
crate::store::event_follow_up::enqueue_push_match(tx.conn(), community, id, kind)
|
|
.await
|
|
.expect("enqueue");
|
|
tx.record_channel_event(channel, kind);
|
|
}
|
|
pub(crate) async fn mention_writer(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO event_mentions (community_id, event_id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
}
|
|
pub async fn verify_floor_guard_behavior(pool: &PgPool) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(pool)
|
|
.await
|
|
.expect("probe");
|
|
}
|
|
pub(crate) async fn lowercase_writer(tx: &mut AdmittedTx) {
|
|
sqlx::query("insert into
|
|
public.events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
}
|
|
pub(crate) async fn continued_writer(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO \
|
|
events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
}
|
|
/// Kept at module scope, not `#[cfg(test)]`, because production calls it.
|
|
pub(crate) async fn writer_under_doc_mention(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
}
|
|
// see `#[cfg(test)] mod x {`
|
|
pub(crate) async fn writer_under_brace_mention(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
}
|
|
pub(crate) async fn writer_with_escaped_newline(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO\nevents (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
}
|
|
pub(crate) async fn writer_with_commented_out_hook(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
// crate::store::event_follow_up::after_admitted_insert(tx, id, kind, channel)
|
|
}
|
|
pub(crate) async fn writer_with_block_commented_hook(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
/* crate::store::event_follow_up::enqueue_push_match(tx.conn(), community, id, kind)
|
|
/* nested */ tx.record_channel_event(channel, kind); */
|
|
}
|
|
pub(crate) async fn writer_with_hook_in_string(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
tracing::debug!("skipped event_follow_up::after_admitted_insert(tx, ..)");
|
|
}
|
|
pub(crate) async fn writer_with_nested_block_commented_hook(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
/* /* inner */ crate::store::event_follow_up::after_admitted_insert(tx, id, kind, channel) */
|
|
}
|
|
pub(crate) async fn writer_with_hook_after_escaped_quote(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
tracing::debug!("\"event_follow_up::after_admitted_insert(\"");
|
|
}
|
|
pub(crate) async fn writer_with_hook_in_raw_string(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
tracing::debug!(r#""event_follow_up::after_admitted_insert(""#);
|
|
}
|
|
/*
|
|
#[cfg(test)]
|
|
*/
|
|
pub(crate) async fn writer_after_block_commented_cfg_test(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
}
|
|
pub(crate) async fn writer_after_raw_string_example(tx: &mut AdmittedTx) {
|
|
let _example = r#"
|
|
fn example() {
|
|
crate::store::event_follow_up::after_admitted_insert(tx, id, kind, channel)
|
|
}
|
|
"#;
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
}
|
|
#[cfg(test)]
|
|
fn test_only_helper() -> (&'static str, char, &'static str, char, &'static [u8]) {
|
|
// an unbalanced { in a comment
|
|
("{", '{', r#"}"} {"#, '"', br#"\"#)
|
|
}
|
|
pub(crate) async fn writer_after_test_helper(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(tx.conn())
|
|
.await
|
|
.expect("write");
|
|
}
|
|
#[cfg(test)]
|
|
mod tests {
|
|
async fn test_writer(pool: &PgPool) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
|
.execute(pool)
|
|
.await
|
|
.expect("fixture");
|
|
}
|
|
}
|
|
"##;
|
|
|
|
assert_eq!(
|
|
event_insert_follow_up_violations(&strip_cfg_test_items(source)),
|
|
[
|
|
"pub(crate) async fn unhooked_writer(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn push_only_writer(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn lowercase_writer(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn continued_writer(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_under_doc_mention(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_under_brace_mention(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_with_escaped_newline(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_with_commented_out_hook(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_with_block_commented_hook(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_with_hook_in_string(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_with_nested_block_commented_hook(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_with_hook_after_escaped_quote(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_with_hook_in_raw_string(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_after_block_commented_cfg_test(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_after_raw_string_example(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_after_test_helper(tx: &mut AdmittedTx) {",
|
|
],
|
|
"an events insert must run the push enqueue and record the TTL refresh"
|
|
);
|
|
}
|
|
|
|
fn production_event_writer_source_files() -> Vec<std::path::PathBuf> {
|
|
use std::path::{Path, PathBuf};
|
|
|
|
fn collect_rs_files(dir: &Path, out: &mut Vec<PathBuf>) {
|
|
for entry in std::fs::read_dir(dir).expect("read source directory") {
|
|
let path = entry.expect("read directory entry").path();
|
|
if path.is_dir() {
|
|
collect_rs_files(&path, out);
|
|
} else if should_scan_guarded_write_source_file(&path) {
|
|
out.push(path);
|
|
}
|
|
}
|
|
}
|
|
|
|
let crates_root = Path::new(env!("CARGO_MANIFEST_DIR"))
|
|
.parent()
|
|
.expect("crate lives under crates/");
|
|
let mut files = Vec::new();
|
|
for entry in std::fs::read_dir(crates_root).expect("read crates directory") {
|
|
let src = entry.expect("read crates entry").path().join("src");
|
|
if src.is_dir() {
|
|
collect_rs_files(&src, &mut files);
|
|
}
|
|
}
|
|
|
|
files
|
|
}
|
|
|
|
#[test]
|
|
fn every_production_event_insert_runs_the_follow_up_hook() {
|
|
let files = production_event_writer_source_files();
|
|
let mut hooked_writers = 0usize;
|
|
let mut violations = Vec::new();
|
|
for path in files {
|
|
let source = std::fs::read_to_string(&path).expect("read source file");
|
|
let production = strip_cfg_test_items(&source);
|
|
if !inserts_events_row(&production) {
|
|
continue;
|
|
}
|
|
for function_source in function_slices(&production) {
|
|
if inserts_events_row(function_source) && runs_event_follow_ups(function_source) {
|
|
hooked_writers += 1;
|
|
}
|
|
}
|
|
violations.extend(
|
|
event_insert_follow_up_violations(&production)
|
|
.into_iter()
|
|
.map(|header| format!("{}: {header}", path.display())),
|
|
);
|
|
}
|
|
|
|
assert!(
|
|
violations.is_empty(),
|
|
"production `INSERT INTO events` writers must call `event_follow_up` because no \
|
|
trigger does the push enqueue or channel TTL refresh: {violations:?}"
|
|
);
|
|
// The seven writers at the time of BUZZ-176. A lower count means the scan
|
|
// stopped seeing a writer, not that one was removed safely.
|
|
assert!(
|
|
hooked_writers >= 7,
|
|
"expected at least seven hooked event writers, found {hooked_writers}"
|
|
);
|
|
}
|
|
|
|
// Fixed channel-less kinds cannot be channel rosters, and the replica-floor
|
|
// startup probe rolls back all its writes.
|
|
// Exemptions match by function header, so a kind 39002 insert added inside one
|
|
// of these functions would pass the policy; keep them single-purpose.
|
|
const EVENT_INSERT_ROSTER_GUARD_EXCEPTIONS: [&str; 3] = [
|
|
// NIP-43 membership is kind 13534 with channel_id NULL.
|
|
"pub(crate) async fn write_nip43_membership_snapshot(",
|
|
// Push leases are kind 30350 with channel_id NULL.
|
|
"pub async fn accept_lease_event(",
|
|
// Rolled-back replica-floor startup probe.
|
|
"pub async fn verify_floor_guard_behavior(",
|
|
];
|
|
|
|
fn event_insert_roster_guard_violations(production_source: &str) -> Vec<String> {
|
|
function_slices(production_source)
|
|
.into_iter()
|
|
.filter(|function_source| {
|
|
let header = function_header(function_source);
|
|
let code = mask_comments_and_literals(function_source);
|
|
inserts_events_row(function_source)
|
|
&& !EVENT_INSERT_ROSTER_GUARD_EXCEPTIONS
|
|
.iter()
|
|
.any(|exception| header.starts_with(exception))
|
|
&& !code.contains("guard_channel_roster_event(")
|
|
&& !code.contains("validate_channel_roster_snapshot(")
|
|
})
|
|
.map(|function_source| function_header(function_source).to_owned())
|
|
.collect()
|
|
}
|
|
|
|
#[test]
|
|
fn event_insert_roster_guard_policy_rejects_writers_without_the_guard() {
|
|
let source = r#"
|
|
pub(crate) async fn unguarded_writer(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
|
|
}
|
|
pub(crate) async fn guarded_writer(tx: &mut AdmittedTx) {
|
|
guard_channel_roster_event(tx, kind, channel, tags).await?;
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
|
|
}
|
|
pub(crate) async fn locked_writer(tx: &mut AdmittedTx) {
|
|
validate_channel_roster_snapshot(tx, channel, tags).await?;
|
|
sqlx::query("insert into public.events (community_id, id) VALUES ($1, $2)");
|
|
}
|
|
pub(crate) async fn mention_writer(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO event_mentions (community_id, event_id) VALUES ($1, $2)");
|
|
}
|
|
pub(crate) async fn commented_guard_writer(tx: &mut AdmittedTx) {
|
|
// guard_channel_roster_event(tx, kind, channel, tags).await?;
|
|
/* validate_channel_roster_snapshot(tx, channel, tags).await?; */
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
|
|
}
|
|
pub(crate) async fn quoted_guard_writer(tx: &mut AdmittedTx) {
|
|
tracing::debug!("guard_channel_roster_event(");
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
|
|
}
|
|
pub(crate) async fn write_nip43_membership_snapshot(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
|
|
}
|
|
pub async fn accept_lease_event(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
|
|
}
|
|
pub async fn verify_floor_guard_behavior(pool: &PgPool) {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
|
|
}
|
|
#[cfg(test)]
|
|
mod tests {
|
|
fn fixture() {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
|
|
}
|
|
}
|
|
"#;
|
|
assert_eq!(
|
|
event_insert_roster_guard_violations(&strip_cfg_test_items(source)),
|
|
[
|
|
"pub(crate) async fn unguarded_writer(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn commented_guard_writer(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn quoted_guard_writer(tx: &mut AdmittedTx) {",
|
|
],
|
|
"only real roster-guard calls in production writers count"
|
|
);
|
|
}
|
|
|
|
/// Presence check: every production `INSERT INTO events` function must contain
|
|
/// a real (uncommented, unquoted) roster-guard call. It does not prove the call
|
|
/// is awaited, reachable, or given the event's real kind and channel; the PG
|
|
/// `app_roster_guard_*` tests cover behavior.
|
|
#[test]
|
|
fn every_production_event_insert_runs_the_roster_guard() {
|
|
let mut checked_writers = 0;
|
|
let mut violations = Vec::new();
|
|
for path in production_event_writer_source_files() {
|
|
let source = std::fs::read_to_string(&path).expect("read source file");
|
|
let production = strip_cfg_test_items(&source);
|
|
checked_writers += function_slices(&production)
|
|
.into_iter()
|
|
.filter(|function| inserts_events_row(function))
|
|
.count();
|
|
violations.extend(
|
|
event_insert_roster_guard_violations(&production)
|
|
.into_iter()
|
|
.map(|header| format!("{}: {header}", path.display())),
|
|
);
|
|
}
|
|
assert!(
|
|
violations.is_empty(),
|
|
"production `INSERT INTO events` writers must run the app-side roster guard: {violations:?}"
|
|
);
|
|
// Five guarded writers plus the three explicitly exempt writers/probes.
|
|
assert!(
|
|
checked_writers >= 8,
|
|
"expected at least eight event writers, found {checked_writers}"
|
|
);
|
|
}
|
|
|
|
// The rolled-back replica-floor probe writes a kind 9 row, which has no `d`.
|
|
const EVENT_INSERT_D_TAG_EXCEPTIONS: [&str; 1] = ["pub async fn verify_floor_guard_behavior("];
|
|
|
|
/// Whether an `INSERT INTO events` tail names `d_tag` in an explicit column
|
|
/// list. An insert without a column list does not count.
|
|
fn events_insert_names_d_tag(tail: &str) -> bool {
|
|
tail.trim_start()
|
|
.strip_prefix('(')
|
|
.and_then(|columns| columns.split_once(')'))
|
|
.is_some_and(|(columns, _)| columns.split(',').any(|column| column.trim() == "d_tag"))
|
|
}
|
|
|
|
fn event_insert_d_tag_violations(production_source: &str) -> Vec<String> {
|
|
function_slices(production_source)
|
|
.into_iter()
|
|
.filter(|function_source| {
|
|
let header = function_header(function_source);
|
|
!EVENT_INSERT_D_TAG_EXCEPTIONS
|
|
.iter()
|
|
.any(|exception| header.starts_with(exception))
|
|
&& !events_insert_tails(function_source)
|
|
.iter()
|
|
.all(|tail| events_insert_names_d_tag(tail))
|
|
})
|
|
.map(|function_source| function_header(function_source).to_owned())
|
|
.collect()
|
|
}
|
|
|
|
#[test]
|
|
fn event_insert_d_tag_policy_rejects_writers_without_the_column() {
|
|
let source = r##"
|
|
pub(crate) async fn writer_without_d_tag(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id, kind) VALUES ($1, $2, $3)");
|
|
}
|
|
pub(crate) async fn writer_with_d_tag(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events \
|
|
(community_id, id, kind, d_tag) VALUES ($1, $2, $3, $4)");
|
|
}
|
|
pub(crate) async fn writer_with_compact_columns(tx: &mut AdmittedTx) {
|
|
sqlx::query("insert into public.events (community_id,id,kind,d_tag) VALUES ($1,$2,$3,$4)");
|
|
}
|
|
pub(crate) async fn writer_with_d_tag_only_in_values(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id, kind) SELECT community_id, id, d_tag FROM x");
|
|
}
|
|
pub(crate) async fn writer_with_lookalike_column(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id, kind, d_tag_hash) VALUES ($1, $2, $3, $4)");
|
|
}
|
|
pub(crate) async fn writer_without_column_list(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events SELECT * FROM staged_events");
|
|
}
|
|
pub(crate) async fn second_insert_without_d_tag(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO events (community_id, id, d_tag) VALUES ($1, $2, $3)");
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
|
|
}
|
|
pub(crate) async fn mention_writer(tx: &mut AdmittedTx) {
|
|
sqlx::query("INSERT INTO event_mentions (community_id, event_id) VALUES ($1, $2)");
|
|
}
|
|
pub async fn verify_floor_guard_behavior(pool: &PgPool) {
|
|
sqlx::query("INSERT INTO events (community_id, id, kind) VALUES ($1, $2, 9)");
|
|
}
|
|
#[cfg(test)]
|
|
mod tests {
|
|
fn fixture() {
|
|
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
|
|
}
|
|
}
|
|
"##;
|
|
assert_eq!(
|
|
event_insert_d_tag_violations(&strip_cfg_test_items(source)),
|
|
[
|
|
"pub(crate) async fn writer_without_d_tag(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_with_d_tag_only_in_values(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_with_lookalike_column(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn writer_without_column_list(tx: &mut AdmittedTx) {",
|
|
"pub(crate) async fn second_insert_without_d_tag(tx: &mut AdmittedTx) {",
|
|
],
|
|
"every production events insert must name `d_tag` in its column list"
|
|
);
|
|
}
|
|
|
|
/// Addressable events (kinds 30000-39999) replace by `d_tag`, and the only
|
|
/// index that serves that lookup is partial on `d_tag IS NOT NULL`. A writer
|
|
/// that leaves the column out stores NULL, so its rows never replace. Nothing
|
|
/// repairs them after the fact, so every production writer must name the column.
|
|
///
|
|
/// This is a text match, not a proof. It checks the column list of literal
|
|
/// `INSERT INTO [public.]events (...)` SQL inside function bodies under each
|
|
/// crate's `src/`. It does not see `ONLY`, quoted or `format!`-built table
|
|
/// names, a comment between the keywords, or SQL held in a constant outside a
|
|
/// function, and it does not check that the bound value is non-NULL. Exempt
|
|
/// probes are skipped by function header, so every insert inside them is too.
|
|
#[test]
|
|
fn every_production_event_insert_names_d_tag() {
|
|
let mut checked_writers = 0;
|
|
let mut violations = Vec::new();
|
|
for path in production_event_writer_source_files() {
|
|
let source = std::fs::read_to_string(&path).expect("read source file");
|
|
let production = strip_cfg_test_items(&source);
|
|
checked_writers += function_slices(&production)
|
|
.into_iter()
|
|
.filter(|function| inserts_events_row(function))
|
|
.count();
|
|
violations.extend(
|
|
event_insert_d_tag_violations(&production)
|
|
.into_iter()
|
|
.map(|header| format!("{}: {header}", path.display())),
|
|
);
|
|
}
|
|
assert!(
|
|
violations.is_empty(),
|
|
"production `INSERT INTO events` writers must name `d_tag` in their column list: \
|
|
{violations:?}"
|
|
);
|
|
// The same writer set as the roster-guard scan. A lower count means the
|
|
// scan stopped seeing a writer, not that one was removed safely.
|
|
assert!(
|
|
checked_writers >= 8,
|
|
"expected at least eight event writers, found {checked_writers}"
|
|
);
|
|
}
|