API — sentry-storage

sentry-storage is Sentry's Postgres persistence: pool, embedded migrations and typed repositories (events, incidents, ip_state, rules, routes, datasets) used by the daemon, the CLI and the dashboard. Every query uses runtime sqlx::query(), so the crate compiles without a live DATABASE_URL. Declares #![forbid(unsafe_code)].

Pool — module pool

PgPool

pub struct PgPool { /* sqlx::PgPool */ }
pub async fn connect(cfg: &PostgresConfig) -> Result<Self>;
pub fn connect_lazy(url: &str) -> Result<Self>;
pub async fn listen(&self, channel: &str) -> Result<sqlx::postgres::PgListener>;
pub async fn notify(&self, channel: &str) -> Result<()>;

PgPool honors max_connections and acquire_timeout_secs from [storage.postgres]; connect_lazy defers connecting (tooling/tests). listen uses a dedicated connection outside the pool — a LISTEN may block indefinitely without starving the pool. Errors live in StorageError (Config/Connect/Query/Migrate).

Repositories — module repo

Repo

Repo holds the pool and lends the repos: events(), incidents(), ip_state(), rules(), routes(), datasets(). Every IP-taking method canonicalizes the address (Event::canonical_ip).

Row types

Each table has its own FromRow/serde struct — these cross the daemon ↔ CLI ↔ dashboard boundary:

EventRepo

Repository for the events table (EventRepo).

Writes and cross-node dedupe:

pub async fn insert_with_hash(&self, evt: &Event, risk_score: u8, risk_level: RiskLevel,
    verdict: Verdict, signals: &serde_json::Value, payload_hash: Option<i64>, process_us: Option<u64>) -> Result<()>;
pub async fn insert_batch_with_hash(&self, rows: &[EventInsert]) -> Result<()>;

insert_with_hash performs a conditional INSERT: when payload_hash was already persisted by a sibling node within the dedupe window (10 s), the row is skipped — the local LRU cannot see other nodes' events (F4.7). insert_batch_with_hash is the overload-response batch persistence lane (one multi-INSERT per batch). EventInsert::from_event builds the row; a protocol that fails to serialize becomes JSON null — batching never loses an event over one bad payload.

Reads and aggregations:

IncidentRepo

Repository for the incidents table (IncidentRepo): create, get_or_create_for_event (idempotent by event_id — partial unique index), open_incident_for_ip (coalescing: one open incident per IP), unresolved, ack and resolve (the round-trip for webhooks carrying an ack_url).

IpStateRepo

Repository for the ip_state table (IpStateRepo) — the source of truth for blocks and the repeat-offender memory:

RuleRepo

Repository for the rules table (RuleRepo): load_ruleset compiles the rows (match_expr via the DSL) into a RuleSet loaded at startup; upsert, delete, list and set_enabled serve the CLI/dashboard.

RouteRepo

Repository for the routes table (RouteRepo): list, insert and delete. RouteRow implements the core's RouteLike trait, so it plugs straight into RouteValidator::merge (config ∪ DB).

DatasetRepo

Repository for DB-backed datasets (F7.7, DatasetRepo) — curated user_agent/path/ja3 lists that become synthetic feed:<name> rules and feed the dynamic prefilter:

pub async fn upsert(&self, name: &str, kind: &str, source_url: Option<&str>,
    action: &str, entries: &[String]) -> Result<()>;
pub async fn list(&self) -> Result<Vec<DatasetRow>>;
pub async fn entries(&self, name: &str) -> Result<Vec<String>>;
pub async fn set_enabled(&self, name: &str, enabled: bool) -> Result<()>;
pub async fn delete(&self, name: &str) -> Result<()>;

upsert replaces a dataset wholesale in one transaction and fires pg_notify on the sentry_datasets_changed channel (constant); set_enabled and delete notify the same way — every node hot-reloads without a restart.

Migrations

Embedded via sqlx::migrate! (migrations::run at startup). Files under crates/sentry-storage/migrations/:

Dedupe and LISTEN/NOTIFY

Cross-node dedupe (F4.7) works like this: every node computes the same payload_hash the local LRU uses (IP + method + path for HTTP, raw hash otherwise); the INSERT checks the partial index events_payload_hash_ts within the 10 s window and skips duplicates persisted by siblings. The index is partial (WHERE payload_hash IS NOT NULL) — events without a hash never pay the lookup.

Hot-reload channels via LISTEN/NOTIFY (producers → LISTEN tasks in the daemon):

The typical producer is the CLI/dashboard: a Postgres mutation + NOTIFY via PgPool::notify. The exception is sentry_datasets_changed, emitted inside the DatasetRepo itself, in the same transaction as the write.

Operational notes

• EventRepo::insert is the single-row wrapper without dedupe (delegates to insert_with_hash with payload_hash = None) — for low-volume paths; the daemon's hot path uses the batch lane.
• Repo is a cheap Clone (just the pool handle); so is every lent repo. There is no per-repo state — Postgres is the source of truth, and local caches (BlockTable, ruleset) are synced through the channels above.
• The error module re-exports StorageError/Result defined in pool.rs; every query failure becomes StorageError::Query with the sqlx error in the text.