diff --git a/crates/didbot-pds/src/provision/events.rs b/crates/didbot-pds/src/provision/events.rs index 5f2fd054..7d96d368 100644 --- a/crates/didbot-pds/src/provision/events.rs +++ b/crates/didbot-pds/src/provision/events.rs @@ -326,18 +326,27 @@ where if !self.announceable(did) { return; } - // A relay undoes a delete by putting back the record its op names, - // so a delete that names none is not sent. - if let Some(op) = ops - .iter() - .find(|op| op.action == crate::subscribe::RepoAction::Delete && op.prev.is_none()) - { - tracing::error!( - did, - path = %op.path, - "a delete names no record it removed; not announcing this commit" - ); - return; + // A relay undoes each op once, against the tree before the commit: a + // commit that names one path twice is not sent, and nor is a delete + // that names no record to put back. + let mut paths = BTreeSet::new(); + for op in &ops { + if !paths.insert(op.path.as_str()) { + tracing::error!( + did, + path = %op.path, + "a commit names one path twice; not announcing it" + ); + return; + } + if op.action == crate::subscribe::RepoAction::Delete && op.prev.is_none() { + tracing::error!( + did, + path = %op.path, + "a delete names no record it removed; not announcing this commit" + ); + return; + } } repos.commit(&RepoCommit { did: did.to_owned(), @@ -358,25 +367,28 @@ where /// `outcomes` are the same length and in the same order — [`RecordStore::apply_batch`] /// guarantees that — so they are zipped rather than the caller having to /// carry the record alongside its own outcome. + /// + /// The firehose frame names each key once instead, with the batch's net + /// change to it in `changes`. A relay undoes a commit against the tree + /// before it, not against any state between the batch's operations, and + /// indigo's `NormalizeOps` refuses a commit that names one path twice. pub(super) fn announce_batch( &self, did: &str, writes: &[BatchOp], outcomes: &[BatchOutcome], previous: &[Option], + changes: &[NetChange], at: &Committed, ) { if self.commits.is_none() && self.repos.is_none() { return; } let repository = &at.repository; + // This server's own stream says what each operation wrote, and names + // what a delete removed off `previous`: what each key held before the + // batch applied, read under the batch's own lock. let mut ops = Vec::with_capacity(writes.len()); - // The firehose distinguishes a create from an update and names what - // an update or a delete replaced; this server's own stream does - // neither. So the two lists are built side by side, off `previous` — - // what each key actually held before the batch applied, read under - // the batch's own lock. - let mut repo_ops = Vec::with_capacity(writes.len()); for ((write, outcome), before) in writes.iter().zip(outcomes).zip(previous) { match outcome { BatchOutcome::Written(written) => { @@ -392,16 +404,12 @@ where return; } }; - let cid = written.cid.to_string(); - // A write that leaves its record as it was is no - // mutation, and a frame lists mutations: indigo's - // `InvertOp` refuses an update whose `prev` is its `cid`. - let unchanged = before.as_deref() == Some(cid.as_str()); - let op = CommitOp::written(write.collection(), &written.rkey, cid, record); - if !unchanged { - repo_ops.push(RepoOp::from_commit_op(&op, before.clone())); - } - ops.push(op); + ops.push(CommitOp::written( + write.collection(), + &written.rkey, + written.cid.to_string(), + record, + )); } BatchOutcome::Deleted => { let BatchOp::Delete { rkey, .. } = write else { @@ -411,12 +419,21 @@ where ); return; }; - let op = CommitOp::deleted(write.collection(), rkey, before.clone()); - repo_ops.push(RepoOp::from_commit_op(&op, before.clone())); - ops.push(op); + ops.push(CommitOp::deleted(write.collection(), rkey, before.clone())); } } } + let repo_ops = changes + .iter() + .filter_map(|change| { + Some(RepoOp { + action: change.action()?, + path: didbot_repo::tree_key(&change.collection, &change.rkey), + cid: change.after.as_ref().map(ToString::to_string), + prev: change.before.clone(), + }) + }) + .collect(); self.announce_repo_commit(did, at, repo_ops); if self.commits.is_some() { self.emit_commit(CommitEvent::new( diff --git a/crates/didbot-pds/src/provision/mod.rs b/crates/didbot-pds/src/provision/mod.rs index 62a8cffb..8093f554 100644 --- a/crates/didbot-pds/src/provision/mod.rs +++ b/crates/didbot-pds/src/provision/mod.rs @@ -590,6 +590,37 @@ fn previous_cid(old: Option<&serde_json::Value>) -> Option { } } +/// What a batch did to one key: the record the key held before the batch, +/// and the one the batch left there. +/// +/// A batch may write one key more than once. Its frame and its write log +/// name each key once, with this net change. +struct NetChange { + collection: String, + rkey: String, + /// The CID of the record the key held before the batch, as + /// [`previous_cid`] names it. + before: Option, + /// The CID of the record the batch left there. + after: Option, +} + +impl NetChange { + /// The change as a firehose action, or `None` when the key holds what it + /// held before the batch. + fn action(&self) -> Option { + use crate::subscribe::RepoAction; + match (&self.before, &self.after) { + (None, None) => None, + (None, Some(_)) => Some(RepoAction::Create), + (Some(_), None) => Some(RepoAction::Delete), + (Some(before), Some(after)) => { + (*before != after.to_string()).then_some(RepoAction::Update) + } + } + } +} + /// Refuses a write whose judgment was about a different record. /// /// `judged` is what the write path read to build the subject a gate judged, @@ -793,10 +824,16 @@ impl PolicyGate for FreezeSyncingGate<'_, S> { } /// What admitting a batch produces: each operation's outcome, the one commit -/// they all landed in, and the CID each operation's key held beforehand — -/// which only the read taken under the store's lock can name, and which the -/// firehose frame needs as its `prev`. -type AppliedBatch = (Vec, Committed, Vec>); +/// they all landed in, or `None` when the batch changed no record, the CID +/// each operation's key held beforehand — which only the read taken under the +/// store's lock can name — and the batch's net change to each key, which is +/// what the firehose frame carries. +type AppliedBatch = ( + Vec, + Option, + Vec>, + Vec, +); /// How much of an upload's body [`JudgedUpload`] waits for before it judges /// the upload again. Each judgment after that waits for what has arrived to @@ -1281,13 +1318,14 @@ pub trait Registry: Send + Sync { /// checks [`crate::records::validate_caller`] and [`crate::records::validate_removal`] /// run for a single write — before any of them touches the store, so a /// batch that was going to fail partway fails whole instead. See - /// [`RecordStore::apply_batch`]. A delete of a record the repository does - /// not hold refuses the whole batch. + /// [`RecordStore::apply_batch`]. A delete of a key that holds no record at + /// that point in the batch refuses the whole batch. /// /// Announced as one [`CommitEvent`] carrying every operation, not one per /// write: a batch is one movement of the head, and a consumer reconciling - /// on the commit should see it as one. Its `subscribeRepos` frame leaves - /// out an operation that left its record as it was. + /// on the commit should see it as one. A batch that changes no record + /// makes no commit. The `subscribeRepos` frame of one that does names each + /// key once, with the batch's net change to it. /// /// This is the account writing with its own credential; see /// [`Self::apply_writes_from`]. diff --git a/crates/didbot-pds/src/provision/registry.rs b/crates/didbot-pds/src/provision/registry.rs index 0e6e944b..b78a284c 100644 --- a/crates/didbot-pds/src/provision/registry.rs +++ b/crates/didbot-pds/src/provision/registry.rs @@ -942,18 +942,33 @@ where .iter() .map(|op| self.existing_record_for_op(did, op)) .collect(); - // A delete needs a record to name as its op's `prev`. The - // reference refuses the batch too: its tree will not remove a - // key it does not hold. - if let Some((BatchOp::Delete { collection, rkey }, _)) = writes - .iter() - .zip(&before) - .find(|(op, old)| matches!(op, BatchOp::Delete { .. }) && old.is_none()) - { - return Err(ProvisionError::Record(RecordError::NothingToDelete { - collection: collection.clone(), - rkey: rkey.clone(), - })); + // A delete needs a record to name as its op's `prev`, so its key + // must hold one at that point in the batch, counting what the ops + // before it did there. The reference's tree refuses the same + // batches: it will not remove a key it does not hold. + let mut holds: BTreeMap<(&str, String), bool> = BTreeMap::new(); + for (op, old) in writes.iter().zip(&before) { + let (rkey, written) = match op { + BatchOp::Create { + collection, rkey, .. + } => match crate::records::resolve_key(collection, rkey.as_deref()) { + Ok(crate::records::ResolvedKey::Fixed(fixed)) => (fixed, true), + // A minted key is new, and no other op can name it. + _ => continue, + }, + BatchOp::Update { rkey, .. } => (rkey.clone(), true), + BatchOp::Delete { rkey, .. } => (rkey.clone(), false), + }; + let held = holds + .entry((op.collection(), rkey.clone())) + .or_insert(old.is_some()); + if !written && !*held { + return Err(ProvisionError::Record(RecordError::NothingToDelete { + collection: op.collection().to_owned(), + rkey, + })); + } + *held = written; } // Every blob the batch names, checked before any of it applies — // all-or-nothing, the same rule the record store applies. See @@ -1012,31 +1027,61 @@ where } } } - // Every key this batch touched, whichever operation touched it. A - // `Create` with no requested `rkey` only learns the key it landed on - // from its own outcome, which is why this zips them rather than - // reading `writes` alone. - let touched: BTreeSet = writes - .iter() - .zip(&outcomes) - .map(|(op, outcome)| match (op, outcome) { + // Each key the batch touched, in the order it first did: the + // record it held before the batch, and the one its last op left + // there. A `Create` with no requested `rkey` only learns the key + // it landed on from its own outcome, which is why this zips them + // rather than reading `writes` alone. + type End<'a> = Option<(&'a serde_json::Value, &'a Cid)>; + let mut order: Vec<(&str, &str)> = Vec::new(); + let mut ends: BTreeMap<(&str, &str), (Option<&serde_json::Value>, End<'_>)> = + BTreeMap::new(); + for ((op, outcome), old) in writes.iter().zip(&outcomes).zip(&before) { + let (rkey, after) = match (op, outcome) { ( - BatchOp::Create { collection, .. } | BatchOp::Update { collection, .. }, + BatchOp::Create { record, .. } | BatchOp::Update { record, .. }, BatchOutcome::Written(written), - ) => didbot_repo::tree_key(collection, &written.rkey), - (BatchOp::Delete { collection, rkey }, _) => { - didbot_repo::tree_key(collection, rkey) - } + ) => (written.rkey.as_str(), Some((record, &written.cid))), + (BatchOp::Delete { rkey, .. }, _) => (rkey.as_str(), None), ( BatchOp::Create { .. } | BatchOp::Update { .. }, BatchOutcome::Deleted, ) => { unreachable!("a create or update never produces a delete outcome") } + }; + let key = (op.collection(), rkey); + ends.entry(key) + .and_modify(|end| end.1 = after) + .or_insert_with(|| { + order.push(key); + (old.as_ref(), after) + }); + } + let changes: Vec = order + .iter() + .map(|key| { + let (old, new) = ends[key]; + NetChange { + collection: key.0.to_owned(), + rkey: key.1.to_owned(), + before: previous_cid(old), + after: new.map(|(_, cid)| cid.clone()), + } }) + .filter(|change| change.action().is_some()) + .collect(); + // A batch that changes no record makes no commit, as an + // unchanged `putRecord` makes none and an empty batch makes none. + if changes.is_empty() { + return Ok((outcomes, None, previous, changes)); + } + let touched: BTreeSet = ends + .keys() + .map(|(collection, rkey)| didbot_repo::tree_key(collection, rkey)) .collect(); let at = self.commit_write(&account, touched)?; - Ok((outcomes, at, previous)) + Ok((outcomes, Some(at), previous, changes)) }) }; @@ -1125,44 +1170,37 @@ where .collect(); let gate = self.freeze_syncing_gate(&account); let admitted = self.write_queue.submit_batch(&gate, &subjects, admit)?; - let (outcomes, at, previous) = admitted.result?; - let logged = - writes - .iter() - .zip(&outcomes) - .zip(&previous) - .filter_map(|((op, outcome), held)| match (op, outcome) { - (BatchOp::Delete { collection, rkey }, _) => held.as_ref().map(|_| { - ( - collection.as_str(), - rkey.as_str(), - crate::write_log::Action::Delete, - None, - ) - }), - (_, BatchOutcome::Written(written)) => { - let action = if held.is_some() { - crate::write_log::Action::Update - } else { - crate::write_log::Action::Create - }; - Some(( - op.collection(), - written.rkey.as_str(), - action, - Some(&written.cid), - )) - } - (_, BatchOutcome::Deleted) => None, - }); + let (outcomes, at, previous, changes) = admitted.result?; + let Some(at) = at else { + tracing::info!( + ops = outcomes.len(), + "the batch changed no record; nothing committed" + ); + return Ok(outcomes); + }; + // One row per key the batch changed, as its frame names them. + let logged = changes.iter().filter_map(|change| { + let action = match change.action()? { + crate::subscribe::RepoAction::Create => crate::write_log::Action::Create, + crate::subscribe::RepoAction::Update => crate::write_log::Action::Update, + crate::subscribe::RepoAction::Delete => crate::write_log::Action::Delete, + }; + Some(( + change.collection.as_str(), + change.rkey.as_str(), + action, + change.after.as_ref(), + )) + }); self.log_writes(did, &at, client_id, admitted.revision, logged); tracing::info!( ops = outcomes.len(), + changed = changes.len(), commit = %at.repository.root(), rev = at.repository.rev(), "applied a batch as one commit" ); - self.announce_batch(did, &writes, &outcomes, &previous, &at); + self.announce_batch(did, &writes, &outcomes, &previous, &changes, &at); Ok(outcomes) } diff --git a/crates/didbot-pds/src/records.rs b/crates/didbot-pds/src/records.rs index 09ca913a..b9555712 100644 --- a/crates/didbot-pds/src/records.rs +++ b/crates/didbot-pds/src/records.rs @@ -247,7 +247,7 @@ pub enum RecordError { /// The key the caller asked for. rkey: String, }, - /// A batch deletes a record that is not there when the batch starts. + /// A batch deletes a key that holds no record at that point in the batch. /// /// A firehose delete op names the record it removed as `prev`, and this /// one has none to name. The whole batch is refused, as the reference diff --git a/crates/didbot-pds/tests/write_log.rs b/crates/didbot-pds/tests/write_log.rs index b657b94b..ed466bc6 100644 --- a/crates/didbot-pds/tests/write_log.rs +++ b/crates/didbot-pds/tests/write_log.rs @@ -212,7 +212,9 @@ fn every_write_names_its_app_and_what_it_did() { ); } -/// A batch is one commit: one row per record it wrote, all at its revision. +/// A batch is one commit: one row per record it changed, all at its revision. +/// A key a batch writes twice is one row, and a batch that changes no record +/// commits nothing, so it is not a row. #[test] fn a_batch_logs_each_record_at_one_revision() { let harness = Harness::new(); @@ -248,6 +250,33 @@ fn a_batch_logs_each_record_at_one_revision() { assert_eq!(seen, [("3knew", Action::Create), ("3kold", Action::Delete)]); assert_eq!(rows[0].rev, rows[1].rev); assert!(rows.iter().all(|row| row.client_id.as_deref() == Some(APP))); + + let write = |rkey: &str, text: &str| BatchOp::Update { + collection: COLLECTION.to_owned(), + rkey: rkey.to_owned(), + record: thing(text), + }; + let delete = |rkey: &str| BatchOp::Delete { + collection: COLLECTION.to_owned(), + rkey: rkey.to_owned(), + }; + let before = harness.log.rows().len(); + for batch in [ + vec![write("3knew", "new")], + vec![write("3kgone", "gone"), delete("3kgone")], + vec![write("3ktwice", "once"), write("3ktwice", "twice")], + ] { + harness + .pds + .apply_writes_from(&harness.did, batch, Stance::Optimistic, None, Some(&app)) + .unwrap(); + } + let rows: Vec = harness.log.rows().split_off(before); + let seen: Vec<(&str, Action)> = rows + .iter() + .map(|row| (row.rkey.as_str(), row.action)) + .collect(); + assert_eq!(seen, [("3ktwice", Action::Create)]); } /// A write policy refused never committed, so it is not a row; one policy diff --git a/crates/didbot/tests/subscribe_repos.rs b/crates/didbot/tests/subscribe_repos.rs index 6f2ea598..2d752aad 100644 --- a/crates/didbot/tests/subscribe_repos.rs +++ b/crates/didbot/tests/subscribe_repos.rs @@ -22,7 +22,7 @@ use didbot::data::mst::{key_height, Store}; use didbot::data::{dag_cbor, Cid, Value}; use didbot::identity::AccountDid; use didbot::key::VerifyingKey; -use didbot::pds::{BatchOp, ProvisionError, RecordError, Stance, Swap}; +use didbot::pds::{content_id, BatchOp, ProvisionError, RecordError, Stance, Swap}; use didbot_repo::SignedCommit; use support::commit::read_commit; use support::mst_invert::Session; @@ -596,6 +596,7 @@ impl Relay { ); let ops = ops_of(body); + assert!(!ops.is_empty(), "commit {seq} carries no ops"); let mut paths = BTreeSet::new(); for op in &ops { assert!( @@ -724,6 +725,33 @@ const ORDERS: [[usize; 3]; 6] = [ [2, 1, 0], ]; +/// A batch op creating [`thing`]`(text)` at [`fixed_tid`]`(index)`. +fn create(index: usize, text: &str) -> BatchOp { + BatchOp::Create { + collection: THING.to_owned(), + rkey: Some(fixed_tid(index)), + record: thing(text), + } +} + +/// A batch op writing [`thing`]`(text)` over the record at +/// [`fixed_tid`]`(index)`. +fn update(index: usize, text: &str) -> BatchOp { + BatchOp::Update { + collection: THING.to_owned(), + rkey: fixed_tid(index), + record: thing(text), + } +} + +/// A batch op deleting the record at [`fixed_tid`]`(index)`. +fn delete(index: usize) -> BatchOp { + BatchOp::Delete { + collection: THING.to_owned(), + rkey: fixed_tid(index), + } +} + /// The property a real relay checks: a commit's own frame is enough, on its /// own, to run its operations backwards from the new root to `prevData` — no /// repository, only the blocks this one frame carries. @@ -941,20 +969,9 @@ async fn a_batchs_own_frame_inverts_to_prev_data_in_every_order() { stack.apply( &did, vec![ - BatchOp::Create { - collection: THING.to_owned(), - rkey: Some(fixed_tid(TO)), - record: thing(&format!("entry {FROM}")), - }, - BatchOp::Update { - collection: THING.to_owned(), - rkey: fixed_tid(BESIDE), - record: thing(&format!("entry {BESIDE}, revised")), - }, - BatchOp::Delete { - collection: THING.to_owned(), - rkey: fixed_tid(FROM), - }, + create(TO, &format!("entry {FROM}")), + update(BESIDE, &format!("entry {BESIDE}, revised")), + delete(FROM), ], ); @@ -1012,34 +1029,27 @@ async fn a_write_that_changes_nothing_makes_no_commit() { ); } -/// A batch op that leaves its record as it was is committed with its batch, -/// as the reference PDS commits it, and left out of the frame's ops: an -/// update naming its own `cid` as `prev` is one indigo's `InvertOp` refuses. -/// A batch of nothing else still commits, and its frame lists no ops. +/// A batch that changes no record makes no commit and sends no frame, the +/// same as an unchanged `putRecord` and an empty batch, and an op that leaves +/// its record as it was is left out of the frame of a batch that changes +/// another. A stale `swapCommit` still refuses a batch that would change +/// nothing. #[tokio::test] -async fn a_batchs_unchanged_record_is_left_out_of_its_frame() { +async fn a_batch_that_changes_no_record_makes_no_commit() { let stack = Stack::start(64).await; let mut consumer = stack.follow(None).await; let did = stack.account("kestrel"); let mut relay = Relay::for_account(&stack, &did); provisioned(&mut consumer, &mut relay).await; - write_at(&stack, &did, &mut consumer, &mut relay, [1, 2]).await; + stack.record_at(&did, Some(&fixed_tid(1)), "entry 1"); + let first = next(&mut consumer).await; + let Some(Value::Link(stale)) = field(&first.body, "commit").cloned() else { + panic!("a commit frame carries no commit link"); + }; + relay.accept(&first); + write_at(&stack, &did, &mut consumer, &mut relay, [2]).await; - stack.apply( - &did, - vec![ - BatchOp::Update { - collection: THING.to_owned(), - rkey: fixed_tid(1), - record: thing("entry 1"), - }, - BatchOp::Create { - collection: THING.to_owned(), - rkey: Some(fixed_tid(3)), - record: thing("entry 3"), - }, - ], - ); + stack.apply(&did, vec![update(1, "entry 1"), create(3, "entry 3")]); let ops = relay.accept(&next(&mut consumer).await); let paths: Vec = ops.into_iter().map(|op| op.path).collect(); assert_eq!( @@ -1048,26 +1058,29 @@ async fn a_batchs_unchanged_record_is_left_out_of_its_frame() { "the frame lists a record the batch left as it was" ); - stack.apply( + stack.apply(&did, vec![update(2, "entry 2")]); + stack.apply(&did, vec![create(5, "entry 5"), delete(5)]); + let refused = stack.registry.apply_writes( &did, - vec![BatchOp::Update { - collection: THING.to_owned(), - rkey: fixed_tid(2), - record: thing("entry 2"), - }], + vec![update(2, "entry 2")], + Stance::default(), + Some(&stale), ); - let ops = relay.accept(&next(&mut consumer).await); assert!( - ops.is_empty(), - "the frame lists a record the batch left as it was" + matches!(refused, Err(ProvisionError::SwapCommitFailed { .. })), + "a stale swapCommit did not refuse the batch: {refused:?}" ); + + // None of them sent a frame: the next one is the next write, straight on + // from the last commit the relay accepted. + write_at(&stack, &did, &mut consumer, &mut relay, [4]).await; } -/// A batch that deletes a record not there when it starts is refused whole, -/// even when an earlier op in the batch creates it: the frame's delete op -/// would have no record to name as `prev`. The reference refuses the first -/// kind too, because its tree will not remove a key it does not hold. -/// Nothing in a refused batch lands, and no frame goes out. +/// A batch whose delete finds no record at its key, counting what the ops +/// before it in the batch did there, is refused whole: its op would have no +/// record to name as `prev`. The reference refuses the same batches, since +/// its tree will not remove a key it does not hold. Nothing in a refused +/// batch lands, and no frame goes out. #[tokio::test] async fn a_batch_deleting_a_record_that_is_not_there_is_refused_whole() { let stack = Stack::start(64).await; @@ -1077,16 +1090,10 @@ async fn a_batch_deleting_a_record_that_is_not_there_is_refused_whole() { provisioned(&mut consumer, &mut relay).await; write_at(&stack, &did, &mut consumer, &mut relay, [1]).await; - let create = |index: usize| BatchOp::Create { - collection: THING.to_owned(), - rkey: Some(fixed_tid(index)), - record: thing(&format!("entry {index}")), - }; - let delete = |index: usize| BatchOp::Delete { - collection: THING.to_owned(), - rkey: fixed_tid(index), - }; - for batch in [vec![create(3), delete(2)], vec![create(3), delete(3)]] { + for batch in [ + vec![create(3, "entry 3"), delete(2)], + vec![delete(1), delete(1)], + ] { let refused = stack .registry .apply_writes(&did, batch, Stance::default(), None); @@ -1097,21 +1104,108 @@ async fn a_batch_deleting_a_record_that_is_not_there_is_refused_whole() { ), "a batch deleting a record that is not there was not refused: {refused:?}" ); - assert_eq!( - stack - .registry - .get_record(&did, THING, &fixed_tid(3)) - .expect("the repository is here"), - None, - "part of a refused batch landed" - ); } + let held = |index| { + stack + .registry + .get_record(&did, THING, &fixed_tid(index)) + .expect("the repository is here") + }; + assert_eq!(held(3), None, "part of a refused batch landed"); + assert!(held(1).is_some(), "part of a refused batch landed"); // No frame went out: the next one is the next write, straight on from the // last commit the relay accepted. write_at(&stack, &did, &mut consumer, &mut relay, [4]).await; } +/// Applies `writes` as one batch, has the relay accept its frame, and +/// answers each op the frame carries as `(action, path, cid, prev)`. +async fn batch_ops( + stack: &Stack, + did: &str, + consumer: &mut ws::Client, + relay: &mut Relay, + writes: Vec, +) -> Vec<(String, String, Option, Option)> { + stack.apply(did, writes); + relay + .accept(&next(consumer).await) + .into_iter() + .map(|op| (op.action, op.path, op.cid, op.prev)) + .collect() +} + +/// A batch that writes one key more than once sends one op for it: the net +/// change from the tree before the batch to the tree after. A relay undoes a +/// commit against the tree before it, and indigo's `NormalizeOps` refuses a +/// commit that names one path twice. Each frame still inverts to `prevData` +/// from its own blocks. +#[tokio::test] +async fn a_batch_writing_one_key_twice_sends_its_net_change() { + let stack = Stack::start(64).await; + let mut consumer = stack.follow(None).await; + let did = stack.account("kestrel"); + let mut relay = Relay::for_account(&stack, &did); + provisioned(&mut consumer, &mut relay).await; + write_at(&stack, &did, &mut consumer, &mut relay, [1, 2]).await; + let cid = |text: &str| Some(content_id(&thing(text)).expect("a record the data model encodes")); + + // A create then an update is a create of the final value. + assert_eq!( + batch_ops( + &stack, + &did, + &mut consumer, + &mut relay, + vec![create(3, "entry 3"), update(3, "entry 3, final")], + ) + .await, + [("create".to_owned(), path(3), cid("entry 3, final"), None)] + ); + // A create then a delete is nothing, beside a write that is something. + assert_eq!( + batch_ops( + &stack, + &did, + &mut consumer, + &mut relay, + vec![create(4, "entry 4"), delete(4), create(5, "entry 5")], + ) + .await, + [("create".to_owned(), path(5), cid("entry 5"), None)] + ); + // An update then a delete is a delete of the record from before the batch. + assert_eq!( + batch_ops( + &stack, + &did, + &mut consumer, + &mut relay, + vec![update(1, "entry 1, gone"), delete(1)], + ) + .await, + [("delete".to_owned(), path(1), None, cid("entry 1"))] + ); + // A delete then a create is an update from the record before the batch. + assert_eq!( + batch_ops( + &stack, + &did, + &mut consumer, + &mut relay, + vec![delete(2), create(2, "entry 2, again")], + ) + .await, + [( + "update".to_owned(), + path(2), + cid("entry 2, again"), + cid("entry 2") + )] + ); +} + /// A relay checks each commit's signature against the `#atproto` key in the /// account's own DID document, and the key another account publishes does /// not verify it.