diff --git a/crates/didbot-pds/src/policy.rs b/crates/didbot-pds/src/policy.rs index f8554cc7..32f0e151 100644 --- a/crates/didbot-pds/src/policy.rs +++ b/crates/didbot-pds/src/policy.rs @@ -21,9 +21,31 @@ //! gate rather than a rule this module encodes: the project owner's answer is //! that different evaluators may need different logic for a delete, so the //! write path's job is only to say, truthfully, which kind of change this is -//! and let whatever is on the other side of [`PolicyGate`] decide what that -//! means. Nothing here treats a delete as "a write to every path", and -//! nothing here special-cases it beyond naming it. +//! and hand over a complete, honest diff — never to decide what a delete +//! means to a field-level rule. +//! +//! # The diff carries values, not just paths +//! +//! An earlier version of this seam carried only changed *paths*. That turned +//! out to be too blunt: a rule that means to test a value can't, from a path +//! alone, tell a legitimate edit from a delete that removed the same path, +//! and a create that under-reports its own paths would bypass a guard a +//! delete is held to. [`Change`] carries both sides' values instead, with +//! absence itself represented as a value (`None`), so an evaluator can test +//! *what changed to what* rather than merely *that something at this path +//! changed*. [`diff`] computes it completely: every path whose value differs +//! between `before` and `after`, including every path in a wholly new record +//! (a create, `before: None` throughout) and every path in a wholly removed +//! one (a delete, `after: None` throughout). There is no "empty diff for a +//! delete" case, and a gate is entitled to rely on that. +//! +//! Nothing here is a payload log. [`Change`] borrows from the two `Value`s a +//! write already holds in memory for the length of one [`PolicyGate::judge`] +//! call, and nothing in this crate stores, logs, or returns a `Change` past +//! that call — see `plan/policy.md`'s "A denial records no part of the +//! refused payload". Transient exposure to an evaluator is what a content +//! rule needs; durable retention is a different thing entirely, and remains +//! this crate's job to keep from happening. use serde_json::Value; @@ -57,12 +79,12 @@ impl std::fmt::Display for PolicyVersion { /// Which kind of change a write makes, named explicitly rather than left for /// a gate to infer from the diff. /// -/// "The diff removed every path" and "the record was deleted" are not the -/// same fact — a record whose every field was cleared by an -/// [`WriteAction::Update`] is still a record — and only an evaluator that can -/// see [`WriteAction::Delete`] directly can tell the two apart. See this -/// module's own docs for why the write path stops there and does not also -/// decide what a delete means to a field-level rule. +/// No longer load-bearing for correctness now that [`Change`] carries values +/// — a gate can tell a removed path from an edited one without being told +/// which kind of write this is — but kept because an evaluator may still +/// want different *logic* per action (the project owner's own framing), and +/// inferring "this was a delete" from "every path lost its `after`" is a +/// fragile inference next to just being told. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum WriteAction { /// The key held nothing before this write. @@ -74,23 +96,159 @@ pub enum WriteAction { Delete, } -/// The content a write is judged on: what a key held, and what it will hold. +/// One field path whose value differs between before and after a write. /// -/// Deliberately dumb: no path-walking, no computed diff. `didbot-policy` -/// decides what "changed paths" means for its own tree (see -/// `plan/policy.md`, "The tree indexes; the list evaluates"); this only hands -/// over the two JSON values a diff would be computed from, alongside the -/// [`WriteAction`] that says which case it is. -#[derive(Debug, Clone, Copy)] -pub struct WriteDiff<'a> { - /// What the key held before this write. `None` for [`WriteAction::Create`]. +/// `before`/`after` are `None` exactly when the path was absent on that +/// side — absence is a value, not a missing [`Change`]. A path present with +/// an identical value on both sides is not represented at all: this is a +/// diff, and an unchanged path did not change. See [`diff`] for how the set +/// of paths is computed, and this module's own docs for why both values are +/// carried rather than the path alone. +/// +/// `path` is an owned `String` rather than a borrow into the record: the +/// paths themselves are computed while walking the structure (`"foo.bar[2]"` +/// is built, not found), and there is nowhere with the right lifetime to +/// borrow one from without leaking it for the life of the process. Only the +/// two `Value`s stay borrowed, and cheaply — they are what an evaluator +/// actually needs to look at. +#[derive(Debug, Clone, PartialEq)] +pub struct Change<'a> { + /// Dot-and-bracket notation from the record's root: `"displayName"`, + /// `"labels.values[0].val"`. + pub path: String, + /// What this path held before the write. `None` if it was absent. pub before: Option<&'a Value>, - /// What the key will hold after this write. `None` for [`WriteAction::Delete`]. + /// What this path holds after the write. `None` if it is now absent. pub after: Option<&'a Value>, } +/// Computes the complete diff between `before` and `after`. +/// +/// Walks both values together, path by path. An object's keys are the union +/// of both sides'; an array is walked by index up to the longer side's +/// length. A path recurses only while both sides at that path are "an object +/// or absent" (respectively "an array or absent"); anywhere the two sides +/// disagree in kind — a scalar replacing an object, an array replacing a +/// scalar, one side absent and the other a leaf — the whole value at that +/// path becomes one [`Change`], because there is no shared substructure left +/// to walk into. An empty object or array on one side, absent on the other, +/// is reported as a single `Change` at its own path rather than as zero +/// changes over zero children — the container itself is the thing that +/// changed. +/// +/// `before: None, after: None` (nothing to diff) produces no changes. +/// `before: None, after: Some(record)` (a create) produces one `Change` per +/// leaf path in `record`, every one with `before: None`. `before: +/// Some(record), after: None` (a delete) is the mirror: one `Change` per leaf +/// path in `record`, every one with `after: None`. There is no case that +/// produces an empty diff for a delete of a non-empty record. +#[must_use] +pub fn diff<'a>(before: Option<&'a Value>, after: Option<&'a Value>) -> Vec> { + let mut changes = Vec::new(); + diff_into("", before, after, &mut changes); + changes +} + +fn join(parent: &str, key: &str) -> String { + if parent.is_empty() { + key.to_owned() + } else { + format!("{parent}.{key}") + } +} + +fn diff_into<'a>(path: &str, before: Option<&'a Value>, after: Option<&'a Value>, out: &mut Vec>) { + if before == after { + return; + } + + let before_object_or_absent = before.is_none() || matches!(before, Some(Value::Object(_))); + let after_object_or_absent = after.is_none() || matches!(after, Some(Value::Object(_))); + if before_object_or_absent && after_object_or_absent { + let mut keys: Vec<&str> = Vec::new(); + if let Some(Value::Object(map)) = before { + keys.extend(map.keys().map(String::as_str)); + } + if let Some(Value::Object(map)) = after { + keys.extend(map.keys().map(String::as_str)); + } + keys.sort_unstable(); + keys.dedup(); + if keys.is_empty() { + // Both sides are objects (or absent) but neither has a key — + // e.g. `{}` on one side and absent on the other. The container + // itself is the change; there is nothing under it to recurse + // into. + out.push(Change { + path: path.to_owned(), + before, + after, + }); + } else { + for key in keys { + let child_before = before.and_then(|value| value.get(key)); + let child_after = after.and_then(|value| value.get(key)); + diff_into(&join(path, key), child_before, child_after, out); + } + } + return; + } + + let before_array_or_absent = before.is_none() || matches!(before, Some(Value::Array(_))); + let after_array_or_absent = after.is_none() || matches!(after, Some(Value::Array(_))); + if before_array_or_absent && after_array_or_absent { + let before_len = before.and_then(Value::as_array).map_or(0, Vec::len); + let after_len = after.and_then(Value::as_array).map_or(0, Vec::len); + let len = before_len.max(after_len); + if len == 0 { + out.push(Change { + path: path.to_owned(), + before, + after, + }); + } else { + for index in 0..len { + let child_before = before.and_then(|value| value.get(index)); + let child_after = after.and_then(|value| value.get(index)); + diff_into(&format!("{path}[{index}]"), child_before, child_after, out); + } + } + return; + } + + // The two sides disagree in kind (object vs. scalar, array vs. object, + // and so on) with neither side absent, or one side is a leaf scalar + // that changed. Either way there is no shared substructure to walk + // into, so the whole value at this path is the change. + out.push(Change { + path: path.to_owned(), + before, + after, + }); +} + +/// Facts about the write's subject, read from this deployment's own records +/// rather than asserted by the caller. +/// +/// A value a caller could assert about itself would be worthless in a +/// policy — an agent could satisfy "must contain the agent's handle" by +/// simply claiming a handle in the write it wants admitted. So every field +/// here is populated by the write path from the account this server already +/// looked up to authorize the write, never from anything in the request. +/// `didbot-policy` is expected to define the canonical attribute set a gate +/// may depend on; this is what the write path can supply today, and it grows +/// as that set does. +#[derive(Debug, Clone, Copy, Default)] +pub struct SubjectAttributes<'a> { + /// The account's own stored handle, if it has one. Read off + /// [`AgentAccount::handle`](crate::account::AgentAccount::handle), which + /// this deployment set at provisioning or a later rename — never off + /// anything in the write being judged. + pub handle: Option<&'a str>, +} + /// One write, as a [`PolicyGate`] judges it: `(agent, client_id, collection, -/// action, diff)`. +/// action, diff, attributes)`. /// /// Per `plan/policy.md`'s "Two trees, one set of mechanics". `client_id` is /// carried even though nothing in this deployment yet issues one to attach @@ -113,8 +271,12 @@ pub struct WriteSubject<'a> { pub collection: &'a str, /// Which kind of change this is. pub action: WriteAction, - /// The content the change is judged on. - pub diff: WriteDiff<'a>, + /// Every path whose value differs between before and after this write, + /// complete and computed by this crate — never trusted from a caller. + pub diff: &'a [Change<'a>], + /// Facts about the agent this deployment can vouch for. See + /// [`SubjectAttributes`]. + pub attributes: SubjectAttributes<'a>, } /// A policy gate's verdict on one write. @@ -186,32 +348,92 @@ impl PolicyGate for NoPolicyGate { #[cfg(test)] mod tests { use super::*; + use serde_json::json; - #[test] - fn an_absent_gate_admits_everything() { - let gate = NoPolicyGate; - let after = serde_json::json!({"text": "hello"}); - let subject = WriteSubject { - agent: "did:plc:test", + fn subject<'a>(agent: &'a str, diff: &'a [Change<'a>]) -> WriteSubject<'a> { + WriteSubject { + agent, client_id: None, collection: "app.bsky.feed.post", action: WriteAction::Create, - diff: WriteDiff { - before: None, - after: Some(&after), - }, - }; - assert_eq!(gate.judge(&subject), Outcome::Allow); + diff, + attributes: SubjectAttributes::default(), + } + } + + #[test] + fn an_absent_gate_admits_everything() { + let gate = NoPolicyGate; + let after = json!({"text": "hello"}); + let changes = diff(None, Some(&after)); + assert_eq!(gate.judge(&subject("did:plc:test", &changes)), Outcome::Allow); assert_eq!(gate.version(), PolicyVersion::none()); } #[test] fn a_pinned_version_survives_a_freeze() { - // Not a behavioural test of `NoPolicyGate` — this just pins the - // shape callers rely on: a version is comparable and displayable, - // because a rejection log line names it. let version = PolicyVersion("v3".to_owned()); assert_eq!(version.to_string(), "v3"); assert_ne!(version, PolicyVersion::none()); } + + #[test] + fn a_create_lists_every_leaf_path_with_no_before() { + let after = json!({"text": "hi", "labels": {"values": [{"val": "x"}]}}); + let changes = diff(None, Some(&after)); + let mut paths: Vec<&str> = changes.iter().map(|c| c.path.as_str()).collect(); + paths.sort_unstable(); + assert_eq!(paths, vec!["labels.values[0].val", "text"]); + assert!(changes.iter().all(|c| c.before.is_none())); + } + + #[test] + fn a_delete_lists_every_leaf_path_with_no_after() { + let before = json!({"text": "hi", "tags": ["a", "b"]}); + let changes = diff(Some(&before), None); + let mut paths: Vec<&str> = changes.iter().map(|c| c.path.as_str()).collect(); + paths.sort_unstable(); + assert_eq!(paths, vec!["tags[0]", "tags[1]", "text"]); + assert!(!changes.is_empty(), "a delete of a non-empty record must never produce an empty diff"); + assert!(changes.iter().all(|c| c.after.is_none())); + } + + #[test] + fn an_update_reports_only_what_actually_changed() { + let before = json!({"text": "hi", "unrelated": "same"}); + let after = json!({"text": "bye", "unrelated": "same"}); + let changes = diff(Some(&before), Some(&after)); + assert_eq!(changes.len(), 1); + assert_eq!(changes[0].path, "text"); + assert_eq!(changes[0].before, Some(&json!("hi"))); + assert_eq!(changes[0].after, Some(&json!("bye"))); + } + + #[test] + fn clearing_a_field_is_a_value_change_not_a_deletion_of_the_record() { + let before = json!({"text": "hi", "displayName": "Bot"}); + let after = json!({"text": "hi", "displayName": ""}); + let changes = diff(Some(&before), Some(&after)); + assert_eq!(changes.len(), 1); + assert_eq!(changes[0].path, "displayName"); + assert_eq!(changes[0].before, Some(&json!("Bot"))); + assert_eq!(changes[0].after, Some(&json!(""))); + } + + #[test] + fn an_empty_object_replacing_absence_is_its_own_change() { + let after = json!({"metadata": {}}); + let changes = diff(None, Some(&after)); + let mut paths: Vec<&str> = changes.iter().map(|c| c.path.as_str()).collect(); + paths.sort_unstable(); + assert_eq!(paths, vec!["metadata"]); + assert_eq!(changes[0].after, Some(&json!({}))); + } + + #[test] + fn nothing_to_diff_produces_no_changes() { + assert!(diff(None, None).is_empty()); + let same = json!({"a": 1}); + assert!(diff(Some(&same), Some(&same)).is_empty()); + } } diff --git a/crates/didbot-pds/src/provision.rs b/crates/didbot-pds/src/provision.rs index ec843f67..6e570c48 100644 --- a/crates/didbot-pds/src/provision.rs +++ b/crates/didbot-pds/src/provision.rs @@ -26,6 +26,7 @@ use crate::kind::{AccountKind, NameProvenance, Retention}; use crate::ledger::{AgentLedger, LedgerEvent, LedgerStore, MemoryLedger}; use crate::lifecycle::{LifecycleEvent, LifecycleSink}; use crate::names::{NameError, Naming}; +use crate::policy::{diff as policy_diff, NoPolicyGate, PolicyGate, SubjectAttributes, WriteAction, WriteSubject}; use crate::records::{ BatchOp, BatchOutcome, ListParams, MemoryRecordStore, Precondition, RecordError, RecordStats, RecordStore, Stance, Written, @@ -325,6 +326,49 @@ pub enum ProvisionError { /// Its current state. state: AccountState, }, + /// A policy refused the write. See [`crate::policy::Outcome::Reject`]. + #[error("write refused by policy: {reason}")] + PolicyRejected { + /// The reason the policy that fired authored. + reason: String, + }, + /// A policy froze the account, refusing this write and every other + /// already waiting for the same repository's turn. See + /// [`crate::policy::Outcome::Freeze`]. + #[error("account frozen by policy: {reason}")] + PolicyFrozen { + /// The reason the freeze was tripped for. + reason: String, + }, + /// More writes were already waiting for this repository's turn against + /// policy than the deployment's queue allows. + /// + /// Not a store failure — the write-ahead log was never touched — and not + /// a policy refusal either: nothing judged this write at all. A caller + /// should back off and retry, the same as + /// [`StoreError::Full`](crate::account::StoreError::Full) asks for a + /// full deployment. + #[error("{did} already has {pending} write(s) waiting for their turn, at the limit of {capacity}")] + WriteQueueFull { + /// The repository whose queue is full. + did: String, + /// How many writes were already waiting. + pending: usize, + /// The configured limit. + capacity: usize, + }, +} + +impl From for ProvisionError { + fn from(error: crate::writequeue::QueueError) -> Self { + match error { + crate::writequeue::QueueError::Full { did, pending, capacity } => { + Self::WriteQueueFull { did, pending, capacity } + } + crate::writequeue::QueueError::Rejected { reason } => Self::PolicyRejected { reason }, + crate::writequeue::QueueError::Frozen { reason } => Self::PolicyFrozen { reason }, + } + } } /// Everything a deployment currently holds, in one snapshot. @@ -1069,6 +1113,16 @@ pub struct Provisioner { /// something: the head a write is checked against cannot move between the /// check and the write, because every path that moves it is here. writing: Mutex<()>, + /// Judges externally-authored writes before they reach `writing`. + /// + /// [`NoPolicyGate`] until a deployment configures otherwise — see + /// [`Self::with_policy_gate`] — which is what "an empty policy set + /// permits everything" means for this field. + policy_gate: Arc, + /// Serializes what `policy_gate` judges, per repository, so one slow + /// judgment never blocks a write to another agent's repository. See + /// [`crate::writequeue`]. + write_queue: crate::writequeue::WriteQueue, /// The repository each DID was last built at, filed under its own root. /// /// Not a second copy of anything: a repository here is derived from the @@ -1186,6 +1240,8 @@ where credentials: Arc::new(MemoryAgentTokenStore::new()), revisions: Minter::new(), writing: Mutex::new(()), + policy_gate: Arc::new(NoPolicyGate), + write_queue: crate::writequeue::WriteQueue::with_default_capacity(), built: Mutex::new(Built::default()), rebuilds: AtomicU64::new(0), sink: None, @@ -1216,6 +1272,27 @@ where self } + /// Judges externally-authored writes with `gate` instead of + /// [`NoPolicyGate`]. + /// + /// The administrative writes enumerated on [`WriteAuthor::Server`] — + /// the registration record and the Bluesky-facing profile, both written + /// at provisioning and again whenever their facts change — never reach + /// `gate` at all; see [`Self::put_record_as`]'s docs for why. + #[must_use] + pub fn with_policy_gate(mut self, gate: Arc) -> Self { + self.policy_gate = gate; + self + } + + /// Bounds how many writes may wait for one repository's turn against + /// `policy_gate`, instead of [`crate::writequeue::DEFAULT_CAPACITY`]. + #[must_use] + pub fn with_write_queue_capacity(mut self, capacity: usize) -> Self { + self.write_queue = crate::writequeue::WriteQueue::new(capacity); + self + } + /// Sends lifecycle events to `sink`. #[must_use] pub fn with_sink(mut self, sink: Arc) -> Self { @@ -2729,46 +2806,97 @@ where // was actually written rather than a re-read that could race a later // write to the same key. let stored = record.clone(); + // Check, write, commit — under one lock, so the head a `swapCommit` // was compared against is still the head the new commit is built on. - let writing = self.writing(); - self.check_commit_for(&account, swap.commit.as_ref())?; - // What this write is about to replace, if the key it resolves to - // already holds something — read before the write, because after it - // the old content is gone and its blob references with it. `None` - // when the key would be minted fresh, which can never collide with - // an existing record. See `records::blob_refs`. - let old = self.existing_record(account.did.as_str(), collection, rkey); - let old_refs = old - .as_ref() - .map(crate::records::blob_refs) - .unwrap_or_default(); - let new_refs = crate::records::blob_refs(&record); - // Named under the same lock and for the same reason: after the write - // nothing downstream of the store can tell a create from an update, - // and the firehose has to say which it was — and, for an update, name - // what was there. A key that resolves to a minted one is always a - // create; nothing holds it yet. - let previous = previous_cid(old.as_ref()); - let written = self - .records - .put(account.did.as_str(), collection, rkey, record, &swap.record) - .inspect_err(|error| { - tracing::info!(%error, "record refused"); - })?; - // Resolved only once the write has actually landed: a refused write - // must not move a reference count that its record never reached. - if !old_refs.is_empty() { - self.blobs - .unmark_referenced(account.did.as_str(), &old_refs); - } - if !new_refs.is_empty() { - self.blobs.mark_referenced(account.did.as_str(), &new_refs); - } - let touched: BTreeSet = - std::iter::once(didbot_repo::tree_key(collection, &written.rkey)).collect(); - let at = self.commit_write(&account, touched)?; - drop(writing); + // Built once and run from two places: directly for a server-authored + // write (see [`WriteAuthor::Server`]'s docs on why those never reach + // a gate), and by [`crate::writequeue::WriteQueue::submit`] for an + // external one, which runs it only once `policy_gate` has returned + // `Allow` — never while holding `self.writing()`, and never at all + // for a write a gate refuses. That is what keeps a denied write from + // ever needing to be un-committed: nothing here runs for it. + let admit = || -> Result<(Written, Committed, Option), ProvisionError> { + let writing = self.writing(); + self.check_commit_for(&account, swap.commit.as_ref())?; + // What this write is about to replace, if the key it resolves to + // already holds something — read before the write, because after + // it the old content is gone and its blob references with it. + // `None` when the key would be minted fresh, which can never + // collide with an existing record. See `records::blob_refs`. + let old = self.existing_record(account.did.as_str(), collection, rkey); + let old_refs = old + .as_ref() + .map(crate::records::blob_refs) + .unwrap_or_default(); + let new_refs = crate::records::blob_refs(&record); + // Named under the same lock and for the same reason: after the + // write nothing downstream of the store can tell a create from + // an update, and the firehose has to say which it was — and, for + // an update, name what was there. A key that resolves to a + // minted one is always a create; nothing holds it yet. + let previous = previous_cid(old.as_ref()); + let written = self + .records + .put(account.did.as_str(), collection, rkey, record.clone(), &swap.record) + .inspect_err(|error| { + tracing::info!(%error, "record refused"); + })?; + // Resolved only once the write has actually landed: a refused + // write must not move a reference count that its record never + // reached. + if !old_refs.is_empty() { + self.blobs + .unmark_referenced(account.did.as_str(), &old_refs); + } + if !new_refs.is_empty() { + self.blobs.mark_referenced(account.did.as_str(), &new_refs); + } + let touched: BTreeSet = + std::iter::once(didbot_repo::tree_key(collection, &written.rkey)).collect(); + let at = self.commit_write(&account, touched)?; + drop(writing); + Ok((written, at, previous)) + }; + + let (written, at, previous) = match author { + // Never reaches `policy_gate` at all — see [`WriteAuthor::Server`] + // and this crate's administrative bypass list in + // [`Self::put_record_as`]'s module-level notes. + WriteAuthor::Server { .. } => admit()?, + WriteAuthor::External => { + // Read once here, outside any lock, purely to build the + // subject a gate judges — `admit` reads the same key again + // under `self.writing()` for the write itself, which is the + // one read this repository's queue guarantees is consistent + // with the commit it accompanies. The two can disagree only + // if something outside this repository's queue changed the + // same key between the two reads — today, only a + // `WriteAuthor::Server` write to the same key, which is rare + // and already narrow: see the bypass list. + let old = self.existing_record(account.did.as_str(), collection, rkey); + let changes = policy_diff(old.as_ref(), Some(&record)); + let action = if old.is_some() { + WriteAction::Update + } else { + WriteAction::Create + }; + let subject = WriteSubject { + agent: account.did.as_str(), + client_id: None, + collection, + action, + diff: &changes, + attributes: SubjectAttributes { + handle: account.handle.as_deref(), + }, + }; + let admitted = self + .write_queue + .submit(self.policy_gate.as_ref(), &subject, admit)?; + admitted.result? + } + }; tracing::Span::current().record("rkey", tracing::field::display(&written.rkey)); // The record's own fields are never logged. A scrobble is a sentence // an agent wrote about its work, and the operator-facing preview of @@ -2800,7 +2928,6 @@ where ); Ok((written, None)) } - /// Refuses a write on an account whose [`AccountState`] does not accept /// one. /// @@ -4044,36 +4171,77 @@ where ) -> Result { let account = self.lookup(did)?; self.require_writable(&account)?; - let _writing = self.writing(); - self.check_commit_for(&account, swap.commit.as_ref())?; - // Read before the removal, for the reason `put_record` reads its own - // old content first: gone is gone, and two things need the old record - // — its blob references, which come off the count with it, and the - // CID it was named by, which is the firehose op's `prev` and the only - // thing on the wire that lets a consumer put it back. + + // Read before any lock, for the reason `put_record_as` reads its own + // old content first: gone is gone, and two things need the old + // record — its blob references, which come off the count with it, + // and the CID it was named by, which is the firehose op's `prev` and + // the only thing on the wire that lets a consumer put it back. Also + // what a policy gate is judged on: the diff of a delete is every + // path this record held, with `after: None` throughout — see + // `crate::policy`'s module docs on why that diff is never empty for + // a record that actually held something. let old = self.records.get(account.did.as_str(), collection, rkey); - let old_refs = old - .as_ref() - .map(crate::records::blob_refs) - .unwrap_or_default(); - let previous = previous_cid(old.as_ref()); - let removed = self - .records - .remove(account.did.as_str(), collection, rkey, &swap.record) - .inspect_err(|error| tracing::info!(%error, "record deletion refused"))?; - if removed && !old_refs.is_empty() { - self.blobs - .unmark_referenced(account.did.as_str(), &old_refs); - } - // Only when something went: a deletion of a record that was not there - // changed no record, and a commit over an unchanged repository would - // move the head for a caller that merely made sure of something. - if removed { - let touched: BTreeSet = - std::iter::once(didbot_repo::tree_key(collection, rkey)).collect(); - let at = self.commit_write(&account, touched)?; - drop(_writing); - self.announce_delete(account.did.as_str(), collection, rkey, &at, previous); + + let admit = || -> Result<(bool, Option, Option), ProvisionError> { + let writing = self.writing(); + self.check_commit_for(&account, swap.commit.as_ref())?; + let old = self.records.get(account.did.as_str(), collection, rkey); + let old_refs = old + .as_ref() + .map(crate::records::blob_refs) + .unwrap_or_default(); + let previous = previous_cid(old.as_ref()); + let removed = self + .records + .remove(account.did.as_str(), collection, rkey, &swap.record) + .inspect_err(|error| tracing::info!(%error, "record deletion refused"))?; + if removed && !old_refs.is_empty() { + self.blobs + .unmark_referenced(account.did.as_str(), &old_refs); + } + // Only when something went: a deletion of a record that was not + // there changed no record, and a commit over an unchanged + // repository would move the head for a caller that merely made + // sure of something. + let at = if removed { + let touched: BTreeSet = + std::iter::once(didbot_repo::tree_key(collection, rkey)).collect(); + Some(self.commit_write(&account, touched)?) + } else { + None + }; + drop(writing); + Ok((removed, at, previous)) + }; + + let (removed, at, previous) = if let Some(old) = old.as_ref() { + // Something is actually being deleted: judge it. A delete of a + // key that holds nothing is a no-op the record store already + // refuses to log; there is no content for a gate to judge and + // routing it through the queue would only cost a turn for + // nothing. + let changes = policy_diff(Some(old), None); + let subject = WriteSubject { + agent: account.did.as_str(), + client_id: None, + collection, + action: WriteAction::Delete, + diff: &changes, + attributes: SubjectAttributes { + handle: account.handle.as_deref(), + }, + }; + let admitted = self + .write_queue + .submit(self.policy_gate.as_ref(), &subject, admit)?; + admitted.result? + } else { + admit()? + }; + + if let Some(at) = &at { + self.announce_delete(account.did.as_str(), collection, rkey, at, previous); } tracing::info!(removed, "record deletion applied"); Ok(removed) diff --git a/crates/didbot-pds/src/writequeue.rs b/crates/didbot-pds/src/writequeue.rs index a5f7a374..b94265cc 100644 --- a/crates/didbot-pds/src/writequeue.rs +++ b/crates/didbot-pds/src/writequeue.rs @@ -360,7 +360,7 @@ impl Drop for Ticket { #[cfg(test)] mod tests { use super::*; - use crate::policy::{PolicyVersion, WriteAction, WriteDiff}; + use crate::policy::{PolicyVersion, SubjectAttributes, WriteAction}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Barrier; use std::time::Duration; @@ -371,10 +371,8 @@ mod tests { client_id: None, collection: "app.bsky.feed.post", action: WriteAction::Create, - diff: WriteDiff { - before: None, - after: None, - }, + diff: &[], + attributes: SubjectAttributes::default(), } }