/// stream events, given positions and broadcast when the batch commits
pub ( crate ) outbox: Outbox ,
/// the pds host each repo row this transaction put counts an account for
pds_accounts: HashMap < Vec < u8 > , Option < SmolStr >> ,
/// heads written or removed in this transaction, so later record ops in the same batch
/// transition counts and history from what earlier ops staged
#[ cfg (feature = "indexer" ) ]
heads: HashMap < Vec < u8 > , Option < fjall:: Slice >> ,
/// latest history death rev staged per record key in this transaction, so two deaths in
/// the same microsecond never clobber each other and a second delete sees the first
#[ cfg (feature = "indexer" ) ]
last_death: HashMap < Vec < u8 > , DbTid > ,
#[ cfg (feature = "indexer" ) ]
lifecycle: Option < LifecycleCountBatch < 'db >> ,
#[ cfg (feature = "indexer" ) ]
record_counts: RecordCounts ,
}
counts,
outbox: Outbox :: default (),
pds_accounts: HashMap :: new (),
#[ cfg (feature = "indexer" ) ]
heads: HashMap :: new (),
#[ cfg (feature = "indexer" ) ]
last_death: HashMap :: new (),
#[ cfg (feature = "indexer" ) ]
lifecycle: None ,
#[ cfg (feature = "indexer" ) ]
record_counts: RecordCounts :: default (),
}
324 unmodified lines
Expand all
);
}
let key = keys:: record_key ( self. did, collection, rkey);
let head = self. txn . db . indexer . head ( & key) . into_diagnostic () ? ;
let head = self. read_head ( & key) ? ;
// counts derive from the actual head transition, never from the
// claimed action: replayed creates, updates of records we never
// saw, and rewrites all count as the storage transition they
Some (_) if head_cid_is_current => {
// the logical version is unchanged, but its representation is
// not: either enforce a durable operator redaction or repair a
// non-redaction CID placeholder once the body is available.
self. txn
. batch
. insert ( &self. txn . db . indexer . heads, key, value);
self. put_head (key, value);
}
Some (old) => {
if !self. only_index_links && !crate ::db:: is_cid_record_value ( & old) {
// a differing head moves to history before being
// replaced, so replays of earlier events can still
// resolve it
self. txn . batch . insert (
&self. txn . db . indexer . history,
keys:: history_key ( & key, &self. death_rev ()),
old . as_ref (),
);
self. write_history ( & key, old . as_ref ( ));
}
self. txn
. batch
. insert ( &self. txn . db . indexer . heads, key, value);
self. put_head (key, value);
}
None => {
self. count_record (collection, 1 );
self. txn
. batch
. insert ( &self. txn . db . indexer . heads, key, value);
self. put_head (key, value);
}
}
if version_is_redacted {
crate ::ops::backlink_ops:: delete_record (
self. txn . counts . add_records (delta);
self. txn . add_record_count ( self. did, collection, delta);
}
fn read_head ( &self , key: & [ u8 ]) -> Result < Option < fjall:: Slice >> {
if let Some (staged) = self. txn . heads . get (key) {
return Ok (staged . clone ());
}
self. txn . db . indexer . head (key) . into_diagnostic ()
}
fn put_head ( &mut self , key: Vec < u8 > , value: & [ u8 ]) {
let value = fjall:: Slice :: from (value);
self. txn
. batch
. insert ( &self. txn . db . indexer . heads, key . clone (), value . clone ());
self. txn . heads . insert (key, Some (value));
}
fn remove_head ( &mut self , key: Vec < u8 > ) {
self. txn . batch . remove ( &self. txn . db . indexer . heads, key . clone ());
self. txn . heads . insert (key, None );
}
fn write_history ( &mut self , key: & [ u8 ], value: & [ u8 ]) {
let death = self. death_rev (key);
self. txn . batch . insert (
&self. txn . db . indexer . history,
keys:: history_key (key, & death),
value,
);
self. txn . last_death . insert (key . to_vec (), death);
}
/// history death key for this commit: the later of the commit rev and
/// the local clock . retention ages history on the same local clock as
/// the local clock , and strictly after any death already staged for `key`
/// in this transaction. retention ages history on the same local clock as
/// event retention, so delayed commits keep their bodies resolvable for
/// the full configured window, and deaths always sort after the events
/// that precede them
fn death_rev ( &self ) -> DbTid {
self. rev . max ( DbTid :: from ( & Tid :: now_0 ()))
fn death_rev ( &self , key: & [ u8 ] ) -> DbTid {
let rev = self. rev . max ( DbTid :: from ( & Tid :: now_0 ())) ;
match self. txn . last_death . get (key) {
Some (last) => rev . max (last . add_micros ( 1 )),
None => rev,
}
}
/// true when a death entry already exists for `key` at or after this
/// commit's rev: this delete (or a later one) has already been applied
fn already_deleted ( &self , key: & [ u8 ]) -> Result < bool > {
if self
. txn
. last_death
. get (key)
. is_some_and ( | death | * death >= self. rev)
{
return Ok ( true );
}
let lo = keys:: history_key (key, &self. rev);
let mut hi = key . to_vec ();
hi . push ( 1 );
Ok ( self
// like put_record, counts derive from the actual head transition:
// deleting a missing record is not a decrement, and replaying a
// delete is a full no-op so the originally preserved body survives
match self. txn . db . indexer . head ( & key) . into_diagnostic () ? {
match self. read_head ( & key) ? {
Some (old) => {
self. count_record (collection, - 1 );
if !self. only_index_links {
// preserve the dying body so replays of earlier events
& []
} else {
old . as_ref ()
};
self. txn . batch . insert (
&self. txn . db . indexer . history,
keys:: history_key ( & key, &self. death_rev ()),
history_value,
);
self. write_history ( & key, history_value);
}
self. txn . batch . remove ( &self. txn . db . indexer . heads, key);
self. remove_head ( key);
}
None => {
if !self. only_index_links && !self. already_deleted ( & key) ? {
// a delete of a record we never saw leaves an empty
// tombstone, distinguishing "deleted at rev" from
// "never existed"
self. txn . batch . insert (
&self. txn . db . indexer . history,
keys:: history_key ( & key, &self. death_rev ()),
[],
);
self. write_history ( & key, & []);
}
}
}
crate ::ops::backlink_ops:: delete_record (
831 unmodified lines
Expand all
assert_eq ! (state . db . indexer . approximate_history_count (), 1 );
}
#[ test ]
fn multiple_writes_in_one_txn_transition_against_staged_heads () {
let (_tmp, state) = test_state ();
let did = did ();
let collection = "app.bsky.feed.post" ;
// 1. create -> update -> delete in one txn (e.g. resync buffer drain)
let mut txn = Txn :: new ( & state . db);
{
let mut r1 = txn . records ( & state, & rev ( "3kzbif5moe22m" ), & did) . unwrap ();
r1 . put_record (
collection,
& DbRkey :: new ( "create-update" ),
&crate ::car:: CarBlock :: from_body ( body ( "v1" )),
DbAction :: Create ,
)
. unwrap ();
r1 . put_record (
collection,
& DbRkey :: new ( "create-delete" ),
&crate ::car:: CarBlock :: from_body ( body ( "ephemeral" )),
DbAction :: Create ,
)
. unwrap ();
r1 . delete_record (collection, & DbRkey :: new ( "never-seen" ))
. unwrap ();
}
{
let mut r2 = txn . records ( & state, & rev ( "3kzbif5mof33m" ), & did) . unwrap ();
r2 . put_record (
collection,
& DbRkey :: new ( "create-update" ),
&crate ::car:: CarBlock :: from_body ( body ( "v2" )),
DbAction :: Update ,
)
. unwrap ();
r2 . delete_record (collection, & DbRkey :: new ( "create-delete" ))
. unwrap ();
r2 . delete_record (collection, & DbRkey :: new ( "create-delete" ))
. unwrap ();
r2 . delete_record (collection, & DbRkey :: new ( "never-seen" ))
. unwrap ();
}
{
let mut r3 = txn . records ( & state, & rev ( "3kzbif5mog44m" ), & did) . unwrap ();
r3 . put_record (
collection,
& DbRkey :: new ( "create-update" ),
&crate ::car:: CarBlock :: from_body ( body ( "v3" )),
DbAction :: Update ,
)
. unwrap ();
}
txn . commit () . unwrap ();
assert_eq ! ( head ( & state, collection, "create-update" ), Some ( b"v3" . to_vec ()));
assert_eq ! (
history_bodies ( & state, collection, "create-update" ),
vec ! [ b"v1" . to_vec (), b"v2" . to_vec ()]
);
assert_eq ! ( head ( & state, collection, "create-delete" ), None );
assert_eq ! (
history_bodies ( & state, collection, "create-delete" ),
vec ! [ b"ephemeral" . to_vec ()]
);
assert_eq ! (
history_bodies ( & state, collection, "never-seen" ),
vec ! [ Vec :: < u8 > :: new ()]
);
assert_eq ! (state . db . get_count_sync ( "records" ), 1 );
assert_eq ! (
crate ::db:: get_record_count ( & state . db, & did, collection) . unwrap (),
1
);
// 2. delete -> recreate in one txn on an already-stored record
let mut txn = Txn :: new ( & state . db);
{
let mut r1 = txn . records ( & state, & rev ( "3kzbif5moh55m" ), & did) . unwrap ();
r1 . delete_record (collection, & DbRkey :: new ( "create-update" ))
. unwrap ();
}
{
let mut r2 = txn . records ( & state, & rev ( "3kzbif5moi66m" ), & did) . unwrap ();
r2 . put_record (
collection,
& DbRkey :: new ( "create-update" ),
&crate ::car:: CarBlock :: from_body ( body ( "v4" )),
DbAction :: Create ,
)
. unwrap ();
}
txn . commit () . unwrap ();
assert_eq ! ( head ( & state, collection, "create-update" ), Some ( b"v4" . to_vec ()));
assert_eq ! (
history_bodies ( & state, collection, "create-update" ),
vec ! [ b"v1" . to_vec (), b"v2" . to_vec (), b"v3" . to_vec ()]
);
assert_eq ! (state . db . get_count_sync ( "records" ), 1 );
assert_eq ! (
crate ::db:: get_record_count ( & state . db, & did, collection) . unwrap (),
1
);
}
#[ test ]
fn unchecked_block_with_a_forged_cid_is_rejected () {
let (_tmp, state) = test_state ();
let mut txn = Txn :: new ( & state . db);
let did = did ();
306 unmodified lines
Expand all