use metering::interval::Sparte;
use sqlx::PgPool;
use time::OffsetDateTime;
use tracing::{info, warn};
use zeroize::Zeroizing;
use crate::error::{Error, Result};
#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct SubjectRef(String);
const EPOCH_PREFIX: char = 's';
pub const MIN_REFERENCE_TOKEN_CHARS: usize = 22;
impl SubjectRef {
pub fn new(reference: impl Into<String>) -> Result<Self> {
let reference = reference.into();
if reference.trim().is_empty() {
return Err(Error::config("subject reference must not be empty"));
}
let this = Self(reference);
this.parts()?;
Ok(this)
}
fn mint(epoch: i32) -> Self {
let mut bytes = [0u8; 16];
getrandom::fill(&mut bytes).expect("OS entropy source unavailable");
Self(format!("{EPOCH_PREFIX}{epoch}_{}", hex(&bytes)))
}
pub fn as_str(&self) -> &str {
&self.0
}
pub fn epoch(&self) -> Result<i32> {
Ok(self.parts()?.0)
}
fn parts(&self) -> Result<(i32, &str)> {
let malformed = |why: &str| {
Error::config(format!(
"subject reference {:?} {why}: the shape is `s<year>_<token>` with a \
token of at least {MIN_REFERENCE_TOKEN_CHARS} characters from \
`A-Z a-z 0-9 . _ -`. The year is what lets a write check a reference \
against the year of the reading it is attached to, and the token is \
what makes it a pseudonym rather than a label. Mint one with \
`SubjectRegistry::register`",
self.0
))
};
let rest = self
.0
.strip_prefix(EPOCH_PREFIX)
.ok_or_else(|| malformed("does not name a retention epoch"))?;
let (year, token) = rest
.split_once('_')
.ok_or_else(|| malformed("does not name a retention epoch"))?;
let year = year
.parse::<i32>()
.map_err(|_| malformed("does not name a retention epoch"))?;
if token.chars().count() < MIN_REFERENCE_TOKEN_CHARS {
return Err(malformed("carries too short a token to be a pseudonym"));
}
if !token
.chars()
.all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
{
return Err(malformed("has a token outside the permitted alphabet"));
}
Ok((year, token))
}
}
impl std::fmt::Display for SubjectRef {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.0)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum ErasureTrigger {
Request,
Retention,
}
impl ErasureTrigger {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Request => "request",
Self::Retention => "retention",
}
}
}
impl std::fmt::Display for ErasureTrigger {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
impl std::str::FromStr for ErasureTrigger {
type Err = Error;
fn from_str(s: &str) -> Result<Self> {
match s {
"request" => Ok(Self::Request),
"retention" => Ok(Self::Retention),
other => Err(Error::config(format!(
"{other:?} is not an erasure trigger: it is `request` for an Article 17 \
erasure or `retention` for the § 60 Abs. 6 sweep"
))),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SuppressionLift {
pub at: OffsetDateTime,
pub actor: String,
pub reason: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ErasureRecord {
pub subject: Option<SubjectRef>,
pub erased_at: OffsetDateTime,
pub reason: String,
pub actor: String,
pub trigger: ErasureTrigger,
pub lifted: Option<SuppressionLift>,
}
impl ErasureRecord {
#[must_use]
pub fn epoch(&self) -> Option<i32> {
self.subject.as_ref().and_then(|s| s.epoch().ok())
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SubjectRegistration {
pub subject: SubjectRef,
pub epoch: i32,
pub registered_at: OffsetDateTime,
}
#[derive(Clone)]
pub struct SubjectRegistry {
pool: PgPool,
erasure_keys: Vec<Zeroizing<Vec<u8>>>,
}
impl std::fmt::Debug for SubjectRegistry {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SubjectRegistry")
.field("suppression", &self.suppresses_reregistration())
.field("erasure_keys", &self.erasure_keys.len())
.finish_non_exhaustive()
}
}
pub const MIN_ERASURE_SECRET_BYTES: usize = 32;
impl SubjectRegistry {
pub fn new(pool: PgPool) -> Self {
Self {
pool,
erasure_keys: Vec::new(),
}
}
pub fn with_erasure_secret(pool: PgPool, secret: &[u8]) -> Result<Self> {
Self::with_erasure_keys(pool, &[secret])
}
pub fn with_erasure_keys(pool: PgPool, keys: &[&[u8]]) -> Result<Self> {
if keys.is_empty() {
return Err(Error::config(
"an erasure key ring must hold at least one key: an empty ring \
enforces nothing, and `SubjectRegistry::new` is how a deployment \
says suppression is off rather than looking as though it were on",
));
}
for (i, key) in keys.iter().enumerate() {
if key.len() < MIN_ERASURE_SECRET_BYTES {
return Err(Error::config(format!(
"erasure key {i} is {} bytes and must be at least \
{MIN_ERASURE_SECRET_BYTES} bytes: a shorter key can be \
brute-forced, and the suppression list would then leak the \
identifiers it exists to forget",
key.len()
)));
}
}
Ok(Self {
pool,
erasure_keys: keys.iter().map(|k| Zeroizing::new(k.to_vec())).collect(),
})
}
#[must_use]
pub fn suppresses_reregistration(&self) -> bool {
!self.erasure_keys.is_empty()
}
#[must_use]
pub fn erasure_key_count(&self) -> usize {
self.erasure_keys.len()
}
fn tombstone(&self, natural_id: &str) -> Option<Vec<u8>> {
Some(mac(self.erasure_keys.first()?, natural_id))
}
fn tombstones(&self, natural_id: &str) -> Vec<Vec<u8>> {
self.erasure_keys
.iter()
.map(|key| mac(key, natural_id))
.collect()
}
async fn suppressed_on(&self, conn: &mut sqlx::PgConnection, natural_id: &str) -> Result<bool> {
let tombstones = self.tombstones(natural_id);
if tombstones.is_empty() {
return Ok(false);
}
sqlx::query_scalar::<_, bool>(
"SELECT EXISTS (SELECT 1 FROM meterstore_erasures WHERE natural_id_hmac = ANY($1))",
)
.bind(&tombstones)
.fetch_one(conn)
.await
.map_err(pg)
}
pub async fn create_tables(&self) -> Result<()> {
sqlx::query(
r#"CREATE TABLE IF NOT EXISTS meterstore_subject_map (
subject_ref TEXT PRIMARY KEY,
natural_id TEXT NOT NULL,
-- The calendar year of the values this row may attribute.
-- `(natural_id, epoch)` rather than `natural_id` alone,
-- because § 60 Abs. 6 runs per value: a subject's 2020
-- readings come due while their 2026 readings are current,
-- and one row covering both could satisfy neither.
epoch INTEGER NOT NULL,
registered_at TIMESTAMPTZ NOT NULL DEFAULT now(),
UNIQUE (natural_id, epoch),
-- The epoch is written twice: as this column, which the
-- sweep selects on, and inside the reference, which the
-- write path parses without a round trip. Two spellings of
-- one fact can disagree, and a row where they did would be
-- swept on one year while refusing readings from the other
-- — with nothing to report it, since both look well formed.
CONSTRAINT meterstore_subject_map_epoch_matches_reference
CHECK (starts_with(subject_ref, 's' || epoch::text || '_'))
)"#,
)
.execute(&self.pool)
.await
.map_err(pg)?;
sqlx::query(
r#"CREATE TABLE IF NOT EXISTS meterstore_erasures (
id BIGSERIAL PRIMARY KEY,
-- The reference whose linkage died. NULL for the one case
-- that names none: an `erase_all` against an identifier no
-- mapping was ever created for, which leaves a suppression
-- tombstone and nothing else. UNIQUE rather than the primary
-- key so those rows can exist at all — PostgreSQL treats
-- NULLs as distinct in a unique index, and a surrogate key
-- keeps `ON CONFLICT (subject_ref)` working for the rest.
subject_ref TEXT UNIQUE,
erased_at TIMESTAMPTZ NOT NULL,
reason TEXT NOT NULL,
actor TEXT NOT NULL,
-- Which duty this discharged: 'request' or 'retention'.
-- `reason` is caller-supplied free text, so without this the
-- trail cannot answer either of the two questions a
-- regulator actually asks — show me the requests you
-- handled, and show me that your retention clock runs.
trigger TEXT NOT NULL,
-- Keyed hash of the erased identifier. NULL when the
-- deployment configured no suppression key, in which case a
-- replayed message can re-register the subject — and NULL
-- again once a suppression has been lifted.
natural_id_hmac BYTEA,
-- Set when the suppression was lifted. Lifting reverses a
-- compliance decision, so it is recorded on the row it
-- concerns rather than left to a log line that rotates away.
lifted_at TIMESTAMPTZ,
lifted_by TEXT,
lift_reason TEXT,
CONSTRAINT meterstore_erasures_lift_is_whole
CHECK (num_nulls(lifted_at, lifted_by, lift_reason) IN (0, 3))
)"#,
)
.execute(&self.pool)
.await
.map_err(pg)?;
self.check_schema().await?;
sqlx::query(
r#"CREATE INDEX IF NOT EXISTS meterstore_subject_map_epoch
ON meterstore_subject_map (epoch)"#,
)
.execute(&self.pool)
.await
.map_err(pg)?;
sqlx::query(
r#"CREATE INDEX IF NOT EXISTS meterstore_erasures_hmac
ON meterstore_erasures (natural_id_hmac)
WHERE natural_id_hmac IS NOT NULL"#,
)
.execute(&self.pool)
.await
.map_err(pg)?;
info!(
suppression = self.suppresses_reregistration(),
keys = self.erasure_key_count(),
"subject registry ready"
);
Ok(())
}
async fn check_schema(&self) -> Result<()> {
for (table, expected) in [
(
"meterstore_subject_map",
&["subject_ref", "natural_id", "epoch", "registered_at"][..],
),
(
"meterstore_erasures",
&[
"id",
"subject_ref",
"erased_at",
"reason",
"actor",
"trigger",
"natural_id_hmac",
"lifted_at",
"lifted_by",
"lift_reason",
][..],
),
] {
let present: Vec<String> = sqlx::query_scalar(
"SELECT attname FROM pg_attribute \
WHERE attrelid = to_regclass($1) AND attnum > 0 AND NOT attisdropped",
)
.bind(table)
.fetch_all(&self.pool)
.await
.map_err(pg)?;
let missing: Vec<&str> = expected
.iter()
.copied()
.filter(|column| !present.iter().any(|p| p == column))
.collect();
if !missing.is_empty() {
return Err(Error::config(format!(
"{table} is missing {missing:?}, so it was created by an earlier \
version of this crate. The registry schema changes in place \
rather than through migrations — the crate is unpublished — so \
drop `meterstore_subject_map` and `meterstore_erasures` and call \
`create_tables` again. Both hold compliance state: the mapping is \
rebuilt by re-registering, and the audit trail is not, so export \
it first if this deployment has erased anything"
)));
}
}
Ok(())
}
pub async fn register(
&self,
natural_id: &str,
at: OffsetDateTime,
sparte: Sparte,
) -> Result<SubjectRef> {
self.register_in_epoch(natural_id, retention_epoch(at, sparte))
.await
}
pub async fn register_in_epoch(&self, natural_id: &str, epoch: i32) -> Result<SubjectRef> {
check_natural_id(natural_id)?;
if let Some(existing) = self.lookup_in_epoch(natural_id, epoch).await? {
return Ok(existing);
}
let mut tx = self.pool.begin().await.map_err(pg)?;
lock_natural_id(&mut tx, natural_id).await?;
if let Some(existing) = lookup_in(&mut tx, natural_id, epoch).await? {
tx.commit().await.map_err(pg)?;
return Ok(existing);
}
if self.suppressed_on(&mut tx, natural_id).await? {
crate::observe::metrics()
.registrations_suppressed
.add(1, &[]);
warn!("registration refused for an erased identifier");
return Err(Error::config(
"this identifier was erased and must not be re-registered: \
registering it would rebuild the link Article 17 destroyed. \
A subject who genuinely returns should arrive under a new \
identifier; if the erasure itself was mistaken, lift it \
explicitly with `lift_suppression`",
));
}
let reference = SubjectRef::mint(epoch);
let inserted = sqlx::query_scalar::<_, String>(
r#"INSERT INTO meterstore_subject_map (subject_ref, natural_id, epoch)
VALUES ($1, $2, $3)
ON CONFLICT (natural_id, epoch) DO UPDATE SET natural_id = EXCLUDED.natural_id
RETURNING subject_ref"#,
)
.bind(reference.as_str())
.bind(natural_id)
.bind(epoch)
.fetch_one(&mut *tx)
.await
.map_err(pg)?;
tx.commit().await.map_err(pg)?;
SubjectRef::new(inserted)
}
pub async fn lookup(
&self,
natural_id: &str,
at: OffsetDateTime,
sparte: Sparte,
) -> Result<Option<SubjectRef>> {
self.lookup_in_epoch(natural_id, retention_epoch(at, sparte))
.await
}
pub async fn lookup_in_epoch(
&self,
natural_id: &str,
epoch: i32,
) -> Result<Option<SubjectRef>> {
let mut conn = self.pool.acquire().await.map_err(pg)?;
lookup_in(&mut conn, natural_id, epoch).await
}
pub async fn registrations(&self, natural_id: &str) -> Result<Vec<SubjectRegistration>> {
let rows = sqlx::query_as::<_, (String, i32, OffsetDateTime)>(
"SELECT subject_ref, epoch, registered_at FROM meterstore_subject_map \
WHERE natural_id = $1 ORDER BY epoch",
)
.bind(natural_id)
.fetch_all(&self.pool)
.await
.map_err(pg)?;
rows.into_iter()
.map(|(reference, epoch, registered_at)| {
Ok(SubjectRegistration {
subject: SubjectRef::new(reference)?,
epoch,
registered_at,
})
})
.collect()
}
pub async fn epochs(&self, natural_id: &str) -> Result<Vec<i32>> {
Ok(self
.registrations(natural_id)
.await?
.into_iter()
.map(|r| r.epoch)
.collect())
}
pub async fn references(&self, natural_id: &str) -> Result<Vec<SubjectRef>> {
Ok(self
.registrations(natural_id)
.await?
.into_iter()
.map(|r| r.subject)
.collect())
}
pub async fn resolvable(
&self,
references: &[String],
) -> Result<std::collections::HashSet<String>> {
if references.is_empty() {
return Ok(std::collections::HashSet::new());
}
let found = sqlx::query_scalar::<_, String>(
"SELECT subject_ref FROM meterstore_subject_map WHERE subject_ref = ANY($1)",
)
.bind(references)
.fetch_all(&self.pool)
.await
.map_err(pg)?;
Ok(found.into_iter().collect())
}
pub async fn resolve(&self, subject: &SubjectRef) -> Result<Option<String>> {
sqlx::query_scalar::<_, String>(
"SELECT natural_id FROM meterstore_subject_map WHERE subject_ref = $1",
)
.bind(subject.as_str())
.fetch_optional(&self.pool)
.await
.map_err(pg)
}
pub async fn erase(
&self,
subject: &SubjectRef,
reason: &str,
actor: &str,
now: OffsetDateTime,
) -> Result<Vec<ErasureRecord>> {
self.erase_triggered_by(
Subject::Reference(subject),
reason,
actor,
now,
ErasureTrigger::Request,
)
.await
}
pub async fn erase_all(
&self,
natural_id: &str,
reason: &str,
actor: &str,
now: OffsetDateTime,
) -> Result<Vec<ErasureRecord>> {
self.erase_triggered_by(
Subject::Natural(natural_id),
reason,
actor,
now,
ErasureTrigger::Request,
)
.await
}
pub(crate) async fn erase_triggered_by(
&self,
subject: Subject<'_>,
reason: &str,
actor: &str,
now: OffsetDateTime,
trigger: ErasureTrigger,
) -> Result<Vec<ErasureRecord>> {
let mut tx = self.pool.begin().await.map_err(pg)?;
let (records, destroyed) = self
.erase_linkage(&mut tx, subject, reason, actor, now, trigger)
.await?;
tx.commit().await.map_err(pg)?;
if destroyed {
crate::observe::metrics()
.subjects_erased
.add(1, &crate::observe::erasure_trigger(trigger));
}
Ok(records)
}
pub async fn erase_in(
&self,
conn: &mut sqlx::PgConnection,
subject: &SubjectRef,
reason: &str,
actor: &str,
now: OffsetDateTime,
) -> Result<Vec<ErasureRecord>> {
self.erase_in_owned(conn, Subject::Reference(subject), reason, actor, now)
.await
}
pub async fn erase_all_in(
&self,
conn: &mut sqlx::PgConnection,
natural_id: &str,
reason: &str,
actor: &str,
now: OffsetDateTime,
) -> Result<Vec<ErasureRecord>> {
self.erase_in_owned(conn, Subject::Natural(natural_id), reason, actor, now)
.await
}
async fn erase_in_owned(
&self,
conn: &mut sqlx::PgConnection,
subject: Subject<'_>,
reason: &str,
actor: &str,
now: OffsetDateTime,
) -> Result<Vec<ErasureRecord>> {
let (records, destroyed) = self
.erase_linkage(conn, subject, reason, actor, now, ErasureTrigger::Request)
.await?;
if destroyed {
crate::observe::metrics()
.subjects_erased
.add(1, &crate::observe::erasure_trigger(ErasureTrigger::Request));
}
Ok(records)
}
async fn erase_linkage(
&self,
conn: &mut sqlx::PgConnection,
subject: Subject<'_>,
reason: &str,
actor: &str,
now: OffsetDateTime,
trigger: ErasureTrigger,
) -> Result<(Vec<ErasureRecord>, bool)> {
if reason.trim().is_empty() {
return Err(Error::config("erasure needs a reason for the audit trail"));
}
let tx = conn;
let natural_id = match subject {
Subject::Natural(id) => {
check_natural_id(id)?;
Some(id.to_string())
}
Subject::Reference(reference) => sqlx::query_scalar::<_, String>(
"SELECT natural_id FROM meterstore_subject_map WHERE subject_ref = $1",
)
.bind(reference.as_str())
.fetch_optional(&mut *tx)
.await
.map_err(pg)?,
};
if let Some(id) = natural_id.as_deref() {
lock_natural_id(&mut *tx, id).await?;
}
let tombstone = natural_id.as_deref().and_then(|id| self.tombstone(id));
let ring = natural_id
.as_deref()
.map(|id| self.tombstones(id))
.unwrap_or_default();
let mut targets: Vec<SubjectRef> = match natural_id.as_deref() {
Some(id) => sqlx::query_scalar::<_, String>(
"SELECT subject_ref FROM meterstore_subject_map WHERE natural_id = $1 \
ORDER BY epoch FOR UPDATE",
)
.bind(id)
.fetch_all(&mut *tx)
.await
.map_err(pg)?
.into_iter()
.map(SubjectRef::new)
.collect::<Result<_>>()?,
None => Vec::new(),
};
if targets.is_empty()
&& let Subject::Reference(reference) = subject
{
targets.push(reference.clone());
}
let mut deleted = 0;
for target in &targets {
deleted += sqlx::query("DELETE FROM meterstore_subject_map WHERE subject_ref = $1")
.bind(target.as_str())
.execute(&mut *tx)
.await
.map_err(pg)?
.rows_affected();
}
let mut records = Vec::with_capacity(targets.len());
for target in targets {
sqlx::query(
r#"INSERT INTO meterstore_erasures
(subject_ref, erased_at, reason, actor, trigger, natural_id_hmac)
VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT (subject_ref) DO UPDATE
SET natural_id_hmac =
COALESCE(meterstore_erasures.natural_id_hmac, EXCLUDED.natural_id_hmac)"#,
)
.bind(target.as_str())
.bind(now)
.bind(reason)
.bind(actor)
.bind(trigger.as_str())
.bind(tombstone.as_deref())
.execute(&mut *tx)
.await
.map_err(pg)?;
records.push(ErasureRecord {
subject: Some(target),
erased_at: now,
reason: reason.to_string(),
actor: actor.to_string(),
trigger,
lifted: None,
});
}
if records.is_empty() {
match &tombstone {
Some(hmac) => {
sqlx::query(
r#"INSERT INTO meterstore_erasures
(subject_ref, erased_at, reason, actor, trigger,
natural_id_hmac)
SELECT NULL, $1, $2, $3, $4, $5
WHERE NOT EXISTS (
SELECT 1 FROM meterstore_erasures
WHERE natural_id_hmac = ANY($6))"#,
)
.bind(now)
.bind(reason)
.bind(actor)
.bind(trigger.as_str())
.bind(hmac.as_slice())
.bind(&ring)
.execute(&mut *tx)
.await
.map_err(pg)?;
warn!(
actor,
"erasure requested for an identifier with no mapping: suppressed \
so it cannot be registered later"
);
records.push(ErasureRecord {
subject: None,
erased_at: now,
reason: reason.to_string(),
actor: actor.to_string(),
trigger,
lifted: None,
});
}
None => warn!(
actor,
"erasure requested for an identifier with no mapping and no \
suppression key configured: nothing was recorded, and a later \
registration of it cannot be refused"
),
}
}
if deleted == 0 {
warn!("erasure requested for a subject with no live linkage");
} else {
info!(epochs = deleted, actor, "subject linkage destroyed");
}
Ok((records, deleted > 0))
}
pub async fn expire_epochs_before(
&self,
cutoff: OffsetDateTime,
reason: &str,
actor: &str,
now: OffsetDateTime,
) -> Result<Vec<ErasureRecord>> {
if reason.trim().is_empty() {
return Err(Error::config("erasure needs a reason for the audit trail"));
}
let due_before = sweep_boundary(cutoff);
let mut tx = self.pool.begin().await.map_err(pg)?;
let due = sqlx::query_scalar::<_, String>(
"DELETE FROM meterstore_subject_map WHERE epoch < $1 RETURNING subject_ref",
)
.bind(due_before)
.fetch_all(&mut *tx)
.await
.map_err(pg)?;
if !due.is_empty() {
sqlx::query(
r#"INSERT INTO meterstore_erasures
(subject_ref, erased_at, reason, actor, trigger, natural_id_hmac)
SELECT reference, $2, $3, $4, 'retention', NULL
FROM unnest($1::text[]) AS reference
ON CONFLICT (subject_ref) DO NOTHING"#,
)
.bind(due.as_slice())
.bind(now)
.bind(reason)
.bind(actor)
.execute(&mut *tx)
.await
.map_err(pg)?;
}
tx.commit().await.map_err(pg)?;
let mut records: Vec<ErasureRecord> = due
.into_iter()
.map(|reference| {
Ok(ErasureRecord {
subject: Some(SubjectRef::new(reference)?),
erased_at: now,
reason: reason.to_string(),
actor: actor.to_string(),
trigger: ErasureTrigger::Retention,
lifted: None,
})
})
.collect::<Result<_>>()?;
records.sort_by(|a, b| {
(a.epoch(), a.subject.as_ref().map(SubjectRef::as_str))
.cmp(&(b.epoch(), b.subject.as_ref().map(SubjectRef::as_str)))
});
if !records.is_empty() {
crate::observe::metrics().subjects_erased.add(
records.len() as u64,
&crate::observe::erasure_trigger(ErasureTrigger::Retention),
);
warn!(
epochs = records.len(),
%cutoff,
"retention sweep destroyed linkages past the statutory ceiling"
);
}
Ok(records)
}
pub async fn is_suppressed(&self, natural_id: &str) -> Result<bool> {
if !self.suppresses_reregistration() {
return Ok(false);
}
let mut conn = self.pool.acquire().await.map_err(pg)?;
self.suppressed_on(&mut conn, natural_id).await
}
pub async fn lift_suppression(
&self,
natural_id: &str,
reason: &str,
actor: &str,
now: OffsetDateTime,
) -> Result<bool> {
if reason.trim().is_empty() {
return Err(Error::config(
"lifting a suppression needs a reason: it reverses a compliance \
action and must not be an anonymous edit",
));
}
let tombstones = self.tombstones(natural_id);
if tombstones.is_empty() {
return Err(Error::config(
"no suppression key is configured, so there is no suppression to lift",
));
}
let lifted = sqlx::query(
r#"UPDATE meterstore_erasures
SET natural_id_hmac = NULL,
lifted_at = $2,
lifted_by = $3,
lift_reason = $4
WHERE natural_id_hmac = ANY($1)"#,
)
.bind(&tombstones)
.bind(now)
.bind(actor)
.bind(reason)
.execute(&self.pool)
.await
.map_err(pg)?
.rows_affected();
if lifted > 0 {
warn!(actor, reason, rows = lifted, "erasure suppression lifted");
}
Ok(lifted > 0)
}
pub async fn is_erased(&self, subject: &SubjectRef) -> Result<bool> {
sqlx::query_scalar::<_, bool>(
"SELECT EXISTS (SELECT 1 FROM meterstore_erasures WHERE subject_ref = $1)",
)
.bind(subject.as_str())
.fetch_one(&self.pool)
.await
.map_err(pg)
}
pub async fn erasures(&self, query: &ErasureQuery) -> Result<Vec<ErasureRecord>> {
query.validate()?;
let rows = sqlx::query_as::<
_,
(
Option<String>,
OffsetDateTime,
String,
String,
String,
Option<OffsetDateTime>,
Option<String>,
Option<String>,
),
>(
r#"SELECT subject_ref, erased_at, reason, actor, trigger,
lifted_at, lifted_by, lift_reason
FROM meterstore_erasures
WHERE ($1::timestamptz IS NULL OR erased_at >= $1)
AND ($2::timestamptz IS NULL OR erased_at < $2)
AND ($3::text IS NULL OR trigger = $3)
ORDER BY erased_at DESC, id DESC
LIMIT $4"#,
)
.bind(query.since)
.bind(query.until)
.bind(query.trigger.map(ErasureTrigger::as_str))
.bind(query.limit)
.fetch_all(&self.pool)
.await
.map_err(pg)?;
rows.into_iter()
.map(
|(
subject,
erased_at,
reason,
actor,
trigger,
lifted_at,
lifted_by,
lift_reason,
)| {
Ok(ErasureRecord {
subject: subject.map(SubjectRef::new).transpose()?,
erased_at,
reason,
actor,
trigger: trigger.parse()?,
lifted: lifted_at.map(|at| SuppressionLift {
at,
actor: lifted_by.unwrap_or_default(),
reason: lift_reason.unwrap_or_default(),
}),
})
},
)
.collect()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ErasureQuery {
since: Option<OffsetDateTime>,
until: Option<OffsetDateTime>,
trigger: Option<ErasureTrigger>,
limit: i64,
}
pub const DEFAULT_ERASURE_LIMIT: i64 = 100;
impl Default for ErasureQuery {
fn default() -> Self {
Self::new()
}
}
impl ErasureQuery {
#[must_use]
pub const fn new() -> Self {
Self {
since: None,
until: None,
trigger: None,
limit: DEFAULT_ERASURE_LIMIT,
}
}
#[must_use]
pub const fn since(mut self, since: OffsetDateTime) -> Self {
self.since = Some(since);
self
}
#[must_use]
pub const fn until(mut self, until: OffsetDateTime) -> Self {
self.until = Some(until);
self
}
#[must_use]
pub const fn trigger(mut self, trigger: ErasureTrigger) -> Self {
self.trigger = Some(trigger);
self
}
#[must_use]
pub const fn limit(mut self, limit: i64) -> Self {
self.limit = limit;
self
}
#[must_use]
pub const fn limit_value(self) -> i64 {
self.limit
}
fn validate(self) -> Result<()> {
if self.limit <= 0 {
return Err(Error::config(format!(
"an erasure query's limit is a row count and must be positive; got {}",
self.limit
)));
}
if let (Some(since), Some(until)) = (self.since, self.until)
&& until <= since
{
return Err(Error::config(format!(
"an erasure query's period is half-open `[since, until)`, so {until} \
does not follow {since}: as written it selects nothing, which reads \
as \"nothing was erased\""
)));
}
Ok(())
}
}
#[derive(Debug, Clone, Copy)]
pub(crate) enum Subject<'a> {
Reference(&'a SubjectRef),
Natural(&'a str),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Retention {
CalendarYears(u32),
Rolling(time::Duration),
}
impl Retention {
#[must_use]
pub fn cutoff(self, now: OffsetDateTime) -> OffsetDateTime {
match self {
Self::CalendarYears(years) => {
let year = metering::calendar::local_year(now)
.saturating_sub(i32::try_from(years).unwrap_or(i32::MAX))
.max(EARLIEST_CUTOFF_YEAR);
metering::calendar::year_start_utc(year)
}
Self::Rolling(window) => now
.checked_sub(window)
.unwrap_or_else(|| metering::calendar::year_start_utc(EARLIEST_CUTOFF_YEAR)),
}
}
}
const EARLIEST_CUTOFF_YEAR: i32 = -9998;
#[must_use]
pub fn retention_epoch(at: OffsetDateTime, sparte: Sparte) -> i32 {
crate::planner::balancing_day(at, sparte).year()
}
fn sweep_boundary(cutoff: OffsetDateTime) -> i32 {
metering::calendar::local_year(cutoff)
}
fn check_natural_id(natural_id: &str) -> Result<()> {
if natural_id.trim().is_empty() {
return Err(Error::config("natural identifier must not be empty"));
}
Ok(())
}
async fn lookup_in(
conn: &mut sqlx::PgConnection,
natural_id: &str,
epoch: i32,
) -> Result<Option<SubjectRef>> {
let found = sqlx::query_scalar::<_, String>(
"SELECT subject_ref FROM meterstore_subject_map WHERE natural_id = $1 AND epoch = $2",
)
.bind(natural_id)
.bind(epoch)
.fetch_optional(conn)
.await
.map_err(pg)?;
found.map(SubjectRef::new).transpose()
}
async fn lock_natural_id(conn: &mut sqlx::PgConnection, natural_id: &str) -> Result<()> {
use sha2::{Digest, Sha256};
let digest = Sha256::digest(natural_id.as_bytes());
let key = i64::from_be_bytes(digest[..8].try_into().expect("SHA-256 gives 32 bytes"));
sqlx::query("SELECT pg_advisory_xact_lock($1)")
.bind(key)
.execute(conn)
.await
.map_err(pg)?;
Ok(())
}
fn mac(key: &[u8], natural_id: &str) -> Vec<u8> {
use hmac::{Hmac, Mac};
use sha2::Sha256;
let mut mac = <Hmac<Sha256>>::new_from_slice(key).expect("HMAC accepts keys of any length");
mac.update(natural_id.as_bytes());
mac.finalize().into_bytes().to_vec()
}
fn hex(bytes: &[u8]) -> String {
use std::fmt::Write;
bytes.iter().fold(String::new(), |mut s, b| {
let _ = write!(s, "{b:02x}");
s
})
}
fn pg(e: sqlx::Error) -> Error {
Error::Storage(e.to_string())
}
#[cfg(test)]
mod tests {
use super::*;
const TOKEN: &str = "9f3c1d2e4b5a60718293a4b5c6";
#[test]
fn a_reference_must_not_be_empty() {
assert!(SubjectRef::new("").is_err());
assert!(SubjectRef::new(" ").is_err());
assert!(SubjectRef::new(format!("s2026_{TOKEN}")).is_ok());
}
#[test]
fn a_reference_must_name_its_retention_epoch() {
assert!(SubjectRef::new(format!("sub_{TOKEN}")).is_err());
assert!(SubjectRef::new(format!("s_{TOKEN}")).is_err());
assert!(SubjectRef::new("s2026_").is_err());
assert!(SubjectRef::new(format!("2026_{TOKEN}")).is_err());
assert_eq!(
SubjectRef::new(format!("s2026_{TOKEN}"))
.unwrap()
.epoch()
.unwrap(),
2026
);
assert_eq!(
SubjectRef::new(format!("s-1_{TOKEN}"))
.unwrap()
.epoch()
.unwrap(),
-1
);
}
#[test]
fn a_reference_whose_token_is_too_short_to_be_a_pseudonym_is_refused() {
assert!(SubjectRef::new("s2026_abc").is_err());
assert!(SubjectRef::new("s2026_4821").is_err());
assert!(SubjectRef::new("s2026_deadbeef").is_err());
let short = "a".repeat(MIN_REFERENCE_TOKEN_CHARS - 1);
let just = "a".repeat(MIN_REFERENCE_TOKEN_CHARS);
assert!(SubjectRef::new(format!("s2026_{short}")).is_err());
assert!(SubjectRef::new(format!("s2026_{just}")).is_ok());
}
#[test]
fn the_encodings_a_foreign_minter_uses_are_all_accepted() {
for token in [
"0123456789abcdef0123456789abcdef", "3f8a1c02-7d4e-4b19-9a3e-15c7d2e6b408", "T3JwaGV1cy1XYXNILUhlcmU", "a.b-c_d.e-f_g.h-i_j.k-l", ] {
assert!(
SubjectRef::new(format!("s2026_{token}")).is_ok(),
"{token} should be a well-formed reference"
);
}
for token in [
"0123456789abcdef 0123456789abcde",
"0123456789abcdef/0123456789abcde",
"0123456789abcdef'0123456789abcde",
] {
assert!(
SubjectRef::new(format!("s2026_{token}")).is_err(),
"{token:?} should be refused"
);
}
}
#[test]
fn every_minted_reference_is_one_new_would_accept() {
for epoch in [-1, 0, 1970, 2026, 9999] {
let minted = SubjectRef::mint(epoch);
assert_eq!(minted.epoch().unwrap(), epoch);
assert!(SubjectRef::new(minted.as_str()).is_ok(), "{minted}");
}
}
#[test]
fn an_erasure_trigger_round_trips_through_its_stored_spelling() {
for trigger in [ErasureTrigger::Request, ErasureTrigger::Retention] {
assert_eq!(trigger.as_str().parse::<ErasureTrigger>().unwrap(), trigger);
assert_eq!(trigger.to_string(), trigger.as_str());
}
assert!("sweep".parse::<ErasureTrigger>().is_err());
}
#[test]
fn an_erasure_query_that_selects_nothing_is_refused_rather_than_empty() {
use time::macros::datetime;
assert!(ErasureQuery::new().limit(0).validate().is_err());
assert!(ErasureQuery::new().limit(-1).validate().is_err());
assert!(
ErasureQuery::new()
.since(datetime!(2026-10-01 0:00 UTC))
.until(datetime!(2026-07-01 0:00 UTC))
.validate()
.is_err()
);
assert!(
ErasureQuery::new()
.since(datetime!(2026-07-01 0:00 UTC))
.until(datetime!(2026-07-01 0:00 UTC))
.validate()
.is_err()
);
assert!(
ErasureQuery::new()
.since(datetime!(2026-07-01 0:00 UTC))
.until(datetime!(2026-10-01 0:00 UTC))
.validate()
.is_ok()
);
assert_eq!(ErasureQuery::new().limit_value(), DEFAULT_ERASURE_LIMIT);
}
#[tokio::test]
async fn a_key_ring_reads_with_every_key_and_writes_with_the_first() {
let current = [1u8; 32];
let retired = [2u8; 32];
let ring = SubjectRegistry::with_erasure_keys(lazy_pool(), &[¤t, &retired]).unwrap();
let only_current = SubjectRegistry::with_erasure_secret(lazy_pool(), ¤t).unwrap();
let only_retired = SubjectRegistry::with_erasure_secret(lazy_pool(), &retired).unwrap();
assert_eq!(ring.erasure_key_count(), 2);
assert_eq!(
ring.tombstone("41373559241"),
only_current.tombstone("41373559241"),
"new tombstones are written under the first key"
);
assert_eq!(
ring.tombstones("41373559241"),
vec![
only_current.tombstone("41373559241").unwrap(),
only_retired.tombstone("41373559241").unwrap(),
],
"and a lookup covers both"
);
}
#[tokio::test]
async fn an_empty_key_ring_is_refused_rather_than_read_as_no_suppression() {
assert!(SubjectRegistry::with_erasure_keys(lazy_pool(), &[]).is_err());
assert!(
SubjectRegistry::with_erasure_keys(lazy_pool(), &[&[1u8; 32][..], &[2u8; 8][..]])
.is_err()
);
}
#[test]
fn generated_references_do_not_repeat() {
let refs: std::collections::HashSet<_> =
(0..1_000).map(|_| SubjectRef::mint(2026)).collect();
assert_eq!(refs.len(), 1_000, "references must be unique");
}
#[test]
fn a_generated_reference_is_not_derived_from_anything() {
assert_ne!(SubjectRef::mint(2026), SubjectRef::mint(2026));
}
#[test]
fn a_minted_reference_carries_the_epoch_it_was_minted_for() {
assert_eq!(SubjectRef::mint(2024).epoch().unwrap(), 2024);
}
#[test]
fn the_retention_epoch_is_the_berlin_year() {
use time::macros::datetime;
for sparte in [Sparte::Strom, Sparte::Gas] {
assert_eq!(
retention_epoch(datetime!(2026-12-31 22:59 UTC), sparte),
2026
);
}
assert_eq!(
retention_epoch(datetime!(2026-12-31 23:00 UTC), Sparte::Strom),
2027
);
}
#[test]
fn a_gas_reading_inside_the_gastag_boundary_belongs_to_the_year_it_is_balanced_in() {
use time::macros::datetime;
let at = datetime!(2026-01-01 0:00 UTC);
assert_eq!(retention_epoch(at, Sparte::Gas), 2025);
assert_eq!(retention_epoch(at, Sparte::Strom), 2026);
let later = datetime!(2026-01-01 6:00 UTC); assert_eq!(retention_epoch(later, Sparte::Gas), 2026);
assert_eq!(retention_epoch(later, Sparte::Strom), 2026);
}
#[test]
fn a_rolling_window_expires_whole_years_and_never_erases_early() {
use time::macros::datetime;
for (now, window_days, due_before) in [
(datetime!(2028-06-01 00:00 UTC), 90i64, 2028),
(datetime!(2028-02-01 00:00 UTC), 90, 2027),
(datetime!(2028-01-15 00:00 UTC), 30, 2027),
(datetime!(2028-06-01 00:00 UTC), 400, 2027),
] {
let cutoff = Retention::Rolling(time::Duration::days(window_days)).cutoff(now);
assert_eq!(
sweep_boundary(cutoff),
due_before,
"Rolling({window_days}d) at {now}"
);
}
for window_days in [1i64, 30, 90, 365, 400, 1_000] {
let now = datetime!(2028-06-15 12:00 UTC);
let cutoff = Retention::Rolling(time::Duration::days(window_days)).cutoff(now);
let newest_swept = metering::calendar::year_start_utc(sweep_boundary(cutoff));
assert!(
newest_swept <= cutoff,
"Rolling({window_days}d) would erase a value newer than the window"
);
}
}
#[test]
fn the_sweep_boundary_is_the_calendar_year_of_the_cutoff() {
use time::macros::datetime;
assert_eq!(sweep_boundary(datetime!(2026-01-01 0:00 UTC)), 2026);
assert_eq!(sweep_boundary(datetime!(2025-12-31 23:00 UTC)), 2026);
assert_eq!(sweep_boundary(datetime!(2025-12-31 22:59 UTC)), 2025);
}
#[test]
fn a_whole_gastag_shares_one_epoch_even_across_new_year() {
use time::{Duration, macros::datetime};
let start = datetime!(2025-12-31 5:00 UTC); let mut at = start;
while at < start + Duration::hours(24) {
assert_eq!(
retention_epoch(at, Sparte::Gas),
2025,
"{at} left the epoch its Gastag is balanced in"
);
at += Duration::minutes(15);
}
}
fn lazy_pool() -> PgPool {
PgPool::connect_lazy("postgresql://unused@localhost/unused").expect("a well-formed URL")
}
#[tokio::test]
async fn the_erasure_key_never_reaches_a_log_line() {
let registry = SubjectRegistry::with_erasure_secret(lazy_pool(), &[0xAB; 32]).unwrap();
let shown = format!("{registry:?}");
assert!(!shown.contains("171"), "{shown}");
assert!(!shown.contains("ab"), "{shown}");
assert!(shown.contains("suppression: true"), "{shown}");
assert!(format!("{:?}", SubjectRegistry::new(lazy_pool())).contains("suppression: false"));
}
#[tokio::test]
async fn a_cloned_registry_keeps_a_key_of_its_own() {
let original = SubjectRegistry::with_erasure_secret(lazy_pool(), &[7; 32]).unwrap();
let derived = original.clone();
let expected = original
.tombstone("41373559241")
.expect("a key is configured");
drop(original);
assert_eq!(derived.tombstone("41373559241"), Some(expected));
}
#[tokio::test]
async fn a_tombstone_is_keyed_rather_than_a_bare_hash() {
let a = SubjectRegistry::with_erasure_secret(lazy_pool(), &[1; 32]).unwrap();
let b = SubjectRegistry::with_erasure_secret(lazy_pool(), &[2; 32]).unwrap();
assert_ne!(a.tombstone("41373559241"), b.tombstone("41373559241"));
assert_eq!(a.tombstone("41373559241"), a.tombstone("41373559241"));
assert!(
SubjectRegistry::new(lazy_pool())
.tombstone("41373559241")
.is_none()
);
}
#[test]
fn the_statutory_ceiling_starts_at_the_end_of_the_collection_year() {
use time::macros::datetime;
let policy = Retention::CalendarYears(3);
assert_eq!(
policy.cutoff(datetime!(2028-01-02 00:00 UTC)),
datetime!(2024-12-31 23:00 UTC)
);
assert_eq!(
policy.cutoff(datetime!(2028-12-31 23:00 UTC)),
datetime!(2025-12-31 23:00 UTC)
);
assert_eq!(
policy.cutoff(datetime!(2028-03-01 00:00 UTC)),
policy.cutoff(datetime!(2028-11-30 00:00 UTC)),
);
}
#[test]
fn a_rolling_window_is_measured_from_the_sweep() {
use time::macros::datetime;
let now = datetime!(2028-06-01 00:00 UTC);
assert_eq!(
Retention::Rolling(time::Duration::days(90)).cutoff(now),
now - time::Duration::days(90)
);
}
#[test]
fn a_period_the_calendar_cannot_express_keeps_everything() {
use time::macros::datetime;
let now = datetime!(2028-06-01 00:00 UTC);
let epoch = datetime!(1970-01-01 00:00 UTC);
for years in [u32::MAX, i32::MAX as u32, 100_000, 12_030, 10_000] {
let cutoff = Retention::CalendarYears(years).cutoff(now);
assert!(
cutoff < epoch,
"CalendarYears({years}) put the cutoff at {cutoff}, which makes \
readings due that are nowhere near the ceiling"
);
}
assert!(Retention::Rolling(time::Duration::MAX).cutoff(now) < epoch);
assert_eq!(
Retention::CalendarYears(3).cutoff(now),
datetime!(2024-12-31 23:00 UTC)
);
}
#[tokio::test]
async fn a_short_erasure_key_is_refused() {
assert!(SubjectRegistry::with_erasure_secret(lazy_pool(), &[0; 31]).is_err());
assert!(SubjectRegistry::with_erasure_secret(lazy_pool(), &[0; 32]).is_ok());
}
#[test]
fn an_erasure_record_carries_no_natural_identifier() {
let record = ErasureRecord {
subject: Some(SubjectRef::mint(2026)),
erased_at: OffsetDateTime::UNIX_EPOCH,
reason: "DSAR-2026-0042".to_string(),
actor: "privacy-team".to_string(),
trigger: ErasureTrigger::Request,
lifted: None,
};
let rendered = format!("{record:?}");
assert!(rendered.contains("s2026_"));
assert!(rendered.contains("DSAR-2026-0042"));
assert!(rendered.contains("Request"));
}
}