diff --git a/book/src/drive/time-range-ttl.md b/book/src/drive/time-range-ttl.md index f367e47d075..0aef75e8337 100644 --- a/book/src/drive/time-range-ttl.md +++ b/book/src/drive/time-range-ttl.md @@ -67,10 +67,13 @@ That single property pays off three times: - `ttl` is an optional key of the `timeRange` map, in seconds, parsed into the transform. It is **not part of the grid identity**: - [`TimeRangeTransform::storage_key`] excludes it, so declaring or - changing a TTL never forks the storage level, and query-side grid + [`TimeRangeTransform::storage_key`] excludes it and query-side grid matching ([`TimeRangeGridSpec`]) continues to compare - `(range, step, phase)` only. + `(range, step, phase)` only. Like the rest of the transform it is + **immutable once the contract is registered**: contract updates reject + any change, including adding or removing a TTL — entries written before + a TTL carry storage flags, and an ephemeral level must stay flagless + (refund-free). - **`ttl ≥ range`.** `$createdAt` is consensus-assigned from block time, so writes only ever target windows containing *now*; this invariant guarantees no bucket that can still receive entries (or serve as the @@ -95,6 +98,14 @@ TTL'd index continues drainage of the oldest expired bucket (start `SystemLimits::max_time_range_ttl_drop_operations_per_write` O(1) drop operations and resuming exactly where the previous write's budget ran out. When nothing is expired, the check is a single bounded range read. +The walkers only *request* the drain; it runs once the state +transition's own batch has applied — one drain per level, however many +indexes share it or how many documents the transition carries — so it +can never remove a subtree a queued removal of the same transition still +targets. Its flat drops are collected and applied as one grovedb batch +(a single root-hash propagation per drain); only the indexed-tree +deletes under ranked levels, which have no batched form, execute +immediately. The operation count of a full bucket scales with its distinct groups, and write volume scales with group volume, so drainage keeps pace roughly one window behind; after a quiet spell the backlog amortizes @@ -151,7 +162,8 @@ A time-range bucket is *not* flat, so the platform drains it prefixes when ranked); 4. the emptied bucket is flat-dropped. -Every step is O(1); the *number* of steps scales with the window's +Every step is O(1) in the subtree it drops (the batched drops share one +root-hash propagation); the *number* of steps scales with the window's distinct groups, and that count is what `SystemLimits::max_time_range_ttl_drop_operations_per_write` bounds. **Every write** into a TTL'd index continues drainage where the previous @@ -186,6 +198,17 @@ there is nothing to refund, which is also where TTL writers collectively pre-pay the drainage described below. Cost estimation routes through the same split, so estimated and actual fees stay in the same class. +Two consequences show up in a fee breakdown. Per byte, an ephemeral +write costs 400 (the ordinary written-byte processing rate, +`storage_processing_credit_per_byte`) + 270 = **670 processing credits**, +against 400 processing + 27,000 storage for an ordinary byte. And because +the ephemeral batch is applied *second*, the ancestor chain it shares +with the standing batch (document-type tree up through the contract +tree, contracts root and root key) is re-opened, re-hashed and re-written +once more per state transition; that second propagation is billed as +ephemeral processing — roughly 0.2–0.25M credits per TTL'd write — in +both the estimate and the actual, so `estimated >= actual` still holds. + Drainage itself and the walkers' TTL bookkeeping reads are **unbilled**: their costs go to scratch accounting, never to the triggering user. That is load-bearing for the `estimated >= actual` fee invariant — the diff --git a/packages/rs-dpp/src/data_contract/document_type/index/time_range.rs b/packages/rs-dpp/src/data_contract/document_type/index/time_range.rs index 8ac03d1c97d..a5e8fcc7253 100644 --- a/packages/rs-dpp/src/data_contract/document_type/index/time_range.rs +++ b/packages/rs-dpp/src/data_contract/document_type/index/time_range.rs @@ -94,9 +94,13 @@ pub struct TimeRangeTransform { /// means entries live forever, exactly as before the key existed. /// /// Deliberately **not part of the grid identity**: [`Self::storage_key`] - /// excludes it, so a TTL never forks the storage level, and query-side - /// grid matching (`TimeRangeGridSpec`) compares `(range, step, phase)` - /// only. Contract validation requires `ttl >= range` (a window still + /// excludes it and query-side grid matching (`TimeRangeGridSpec`) + /// compares `(range, step, phase)` only. Like the rest of the transform + /// it is nonetheless immutable once the contract is registered: contract + /// updates reject any change, including adding or removing a TTL, + /// because entries written before a TTL carry storage flags and an + /// ephemeral level must stay flagless (refund-free). Contract validation + /// requires `ttl >= range` (a window still /// able to receive consensus-timestamped writes, or to serve as the /// `oldest` selector's window, can never expire) and caps it at /// `SystemLimits::max_time_range_ttl_seconds` — the cap is what makes diff --git a/packages/rs-dpp/src/data_contract/document_type/index_level/find_first_change.rs b/packages/rs-dpp/src/data_contract/document_type/index_level/find_first_change.rs index 1082ea5d7d4..44bce25c9d8 100644 --- a/packages/rs-dpp/src/data_contract/document_type/index_level/find_first_change.rs +++ b/packages/rs-dpp/src/data_contract/document_type/index_level/find_first_change.rs @@ -197,6 +197,9 @@ impl IndexLevel { /// dictates how many index entries each document produces and under /// which bucket keys, so changing it after creation would leave already /// stored documents indexed under stale buckets — it is immutable. + /// That includes `ttl`, which the storage key leaves out: entries + /// written before a TTL carry storage flags, and an ephemeral level + /// must stay flagless. /// /// Returns `None` if the transform is the same everywhere. #[cfg(feature = "validation")] @@ -204,8 +207,14 @@ impl IndexLevel { if self.time_range() != new.time_range() { let fmt = |t: Option<&super::TimeRangeTransform>| match t { Some(t) => format!( - "Some(on: {:?}, range: {}s, step: {}s, phase: {}s)", - t.source, t.range_seconds, t.step_seconds, t.phase_seconds + "Some(on: {:?}, range: {}s, step: {}s, phase: {}s, ttl: {})", + t.source, + t.range_seconds, + t.step_seconds, + t.phase_seconds, + t.ttl_seconds + .map(|ttl| format!("{}s", ttl)) + .unwrap_or_else(|| "None".to_string()), ), None => "None".to_string(), }; diff --git a/packages/rs-dpp/src/data_contract/document_type/index_level/mod.rs b/packages/rs-dpp/src/data_contract/document_type/index_level/mod.rs index cac5dca3df3..6fe76760866 100644 --- a/packages/rs-dpp/src/data_contract/document_type/index_level/mod.rs +++ b/packages/rs-dpp/src/data_contract/document_type/index_level/mod.rs @@ -2159,4 +2159,84 @@ mod tests { ); } } + + /// A time-range `ttl` is immutable across contract updates even though + /// it is not part of the grid identity — old and new resolve to the + /// same `storage_key` level, so the diff helper must compare the whole + /// transform. Entries written before a TTL carry storage flags, and an + /// ephemeral level must stay flagless. + #[test] + fn should_return_invalid_result_if_time_range_ttl_changed() { + let platform_version = PlatformVersion::latest(); + let document_type_name = "test"; + + let index_with_ttl = |ttl_seconds: Option| Index { + name: "test".to_string(), + properties: vec![IndexProperty { + name: "$createdAt".to_string(), + ascending: false, + }], + unique: false, + null_searchable: true, + contested_index: None, + countable: IndexCountability::NotCountable, + range_countable: false, + summable: None, + range_summable: false, + ranked_countable: false, + ranked_countable_at: vec![], + ranked_summable: false, + ranked_averageable: false, + time_range: Some(TimeRangeTransform { + source: "$createdAt".to_string(), + range_seconds: 3600, + step_seconds: 3600, + phase_seconds: 0, + ttl_seconds, + }), + terminal: None, + preallocated: false, + skip_if_absent: false, + }; + let fmt_ttl = |ttl: Option| { + ttl.map(|ttl| format!("{}s", ttl)) + .unwrap_or_else(|| "None".to_string()) + }; + + for (old_ttl, new_ttl) in [ + (None, Some(86_400)), + (Some(86_400), None), + (Some(86_400), Some(172_800)), + ] { + let old_index_structure = IndexLevel::try_from_indices( + &[index_with_ttl(old_ttl)], + document_type_name, + platform_version, + ) + .expect("failed to create old index level"); + let new_index_structure = IndexLevel::try_from_indices( + &[index_with_ttl(new_ttl)], + document_type_name, + platform_version, + ) + .expect("failed to create new index level"); + + let result = + old_index_structure.validate_update(document_type_name, &new_index_structure); + + let expected_path = format!( + "$createdAt#3600#3600 -> (timeRange: \ + Some(on: \"$createdAt\", range: 3600s, step: 3600s, phase: 0s, ttl: {}) -> \ + Some(on: \"$createdAt\", range: 3600s, step: 3600s, phase: 0s, ttl: {}))", + fmt_ttl(old_ttl), + fmt_ttl(new_ttl), + ); + assert_matches!( + result.errors.as_slice(), + [ConsensusError::BasicError( + BasicError::DataContractInvalidIndexDefinitionUpdateError(e) + )] if e.index_path() == expected_path + ); + } + } } diff --git a/packages/rs-drive-abci/src/abci/handler/finalize_block.rs b/packages/rs-drive-abci/src/abci/handler/finalize_block.rs index f79404a14b7..bc23bf09103 100644 --- a/packages/rs-drive-abci/src/abci/handler/finalize_block.rs +++ b/packages/rs-drive-abci/src/abci/handler/finalize_block.rs @@ -110,10 +110,21 @@ where .flush_pending_prefix_drops(&platform_version.drive.grove_version) { Ok(report) => { - if report.reclaimed_records > 0 || report.skipped_live > 0 { - tracing::debug!( + // A skipped record means a dropped path was re-created before + // its reclamation ran — a violation of the flat-drop path-reuse + // contract that leaks the old prefix data until the path is + // dropped again, so it is worth a warning rather than a debug + // line. + if report.skipped_live > 0 { + tracing::warn!( reclaimed_records = report.reclaimed_records, skipped_live = report.skipped_live, + "flushed pending prefix drops; some dropped paths were live again and \ + were skipped" + ); + } else if report.reclaimed_records > 0 { + tracing::debug!( + reclaimed_records = report.reclaimed_records, "flushed pending prefix drops" ); } diff --git a/packages/rs-drive/src/drive/document/delete/remove_indices_for_top_index_level_for_contract_operations/v2/mod.rs b/packages/rs-drive/src/drive/document/delete/remove_indices_for_top_index_level_for_contract_operations/v2/mod.rs index f2fd88d5bfe..9b58c2113b2 100644 --- a/packages/rs-drive/src/drive/document/delete/remove_indices_for_top_index_level_for_contract_operations/v2/mod.rs +++ b/packages/rs-drive/src/drive/document/delete/remove_indices_for_top_index_level_for_contract_operations/v2/mod.rs @@ -11,7 +11,9 @@ use crate::drive::document::estimation_costs::estimated_sum_trees_for_value_tree use crate::drive::document::index_level_tree_types::{ index_level_tree_types_with_continuation_demotion, time_range_index_keys, }; -use crate::drive::document::time_range_ttl::entry_key_bucket_start; +use crate::drive::document::time_range_ttl::{ + request_expired_time_range_drains, TimeRangeEntryState, +}; use crate::drive::document::unique_event_id; use crate::util::type_constants::DEFAULT_HASH_SIZE_U8; @@ -91,21 +93,19 @@ impl Drive { // TTL drainage rides every write into a TTL'd index — deletes // included: without this, an index receiving only deletions would // never advance cleanup, breaking the documented every-write rule. - // One sweep over the deduplicated levels, BEFORE any delete - // mutation is queued (drainage applies directly to grovedb, so a - // later drain could remove a path a queued operation targets), and - // before the expired/standing detection below so it sees post-drain - // state. Stateful only — the estimation dry run neither reads state - // nor prices drops. Unbilled — see the ttl module's Billing - // section. + // One request per deduplicated level, run once the transition's + // batch has applied (see `apply_batch_low_level_drive_operations`), + // so no drop can race the removals queued below. Stateful only — + // the estimation dry run neither reads state nor prices drops. + // Unbilled — see the ttl module's Billing section. if estimated_costs_only_with_layer_info.is_none() { - self.drain_expired_time_range_levels( + request_expired_time_range_drains( index_level, &contract_document_type_path, block_time_ms, - transaction, platform_version, - )?; + batch_operations, + ); } let sub_level_index_count = index_level.sub_levels().len() as u32; @@ -144,7 +144,6 @@ impl Drive { let sub_level_is_ephemeral = sub_level .time_range() .is_some_and(|transform| transform.ttl_seconds.is_some()); - let mut ephemeral_local_operations: Vec = vec![]; let index_storage_flags = if sub_level_is_ephemeral { None } else { @@ -257,98 +256,92 @@ impl Drive { ); let bucket_count = index_keys.len(); - for (bucket, index_key) in index_keys.into_iter().enumerate() { - // TTL: an expired bucket may already have been dropped - // entirely (this document's entries went with it — skip), - // or stand PARTIALLY drained (drainage removes whole `[0]` - // and group value trees before the bucket): removal then - // proceeds, but at full-path granularity — the deeper - // walkers skip any entry whose path the drain already - // took. Live buckets behave exactly as before. Stateful - // reads have no place in the estimation dry run, which - // processes every bucket — the upper bound. - let mut skip_missing_expired_entry = false; - if estimated_costs_only_with_layer_info.is_none() { - if let Some(transform) = sub_level.time_range() { - let entry_key_bytes = match &index_key { - DriveKeyInfo::Key(key) => Some(key.as_slice()), - DriveKeyInfo::KeyRef(key) => Some(*key), - DriveKeyInfo::KeySize(_) => None, - }; - if let Some(entry_key_bytes) = entry_key_bytes { - let expired = entry_key_bucket_start(entry_key_bytes) - .zip(transform.expiry_horizon_ms(block_time_ms)) - .is_some_and(|(start, horizon)| start < horizon); - if expired { - if !self.time_range_entry_is_removable( - transform, - entry_key_bytes, - block_time_ms, - &index_path, - transaction, - platform_version, - )? { - continue; + LowLevelDriveOperation::with_ephemeral_routing( + batch_operations, + sub_level_is_ephemeral, + |index_batch_operations| { + for (bucket, index_key) in index_keys.into_iter().enumerate() { + // TTL: an expired bucket may already have been + // dropped entirely (this document's entries went + // with it — skip), or stand PARTIALLY drained + // (drainage removes whole `[0]` and group value + // trees before the bucket): removal then proceeds, + // but at full-path granularity — the deeper walkers + // skip any entry whose path the drain already took. + // Live buckets behave exactly as before. Stateful + // reads have no place in the estimation dry run, + // which processes every bucket — the upper bound. + let mut skip_missing_expired_entry = false; + if estimated_costs_only_with_layer_info.is_none() { + if let Some(transform) = sub_level.time_range() { + let entry_key_bytes = match &index_key { + DriveKeyInfo::Key(key) => Some(key.as_slice()), + DriveKeyInfo::KeyRef(key) => Some(*key), + DriveKeyInfo::KeySize(_) => None, + }; + if let Some(entry_key_bytes) = entry_key_bytes { + match self.time_range_entry_state( + transform, + entry_key_bytes, + block_time_ms, + &index_path, + transaction, + platform_version, + )? { + TimeRangeEntryState::Live => {} + TimeRangeEntryState::ExpiredGone => continue, + TimeRangeEntryState::ExpiredStanding => { + skip_missing_expired_entry = true + } + } } - skip_missing_expired_entry = true; } } - } - } - // The final bucket takes ownership of `index_path`; earlier - // buckets (only a time-range fan-out has more than one) - // clone it. - let own_index_path = if bucket + 1 == bucket_count { - std::mem::take(&mut index_path) - } else { - index_path.clone() - }; - let mut index_path_info = if document_and_contract_info - .owned_document_info - .document_info - .is_document_size() - { - // This is a stateless operation - PathInfo::PathWithSizes(KeyInfoPath::from_known_owned_path(own_index_path)) - } else { - PathInfo::PathAsVec::<0>(own_index_path) - }; - - // we push the actual value of the index path - index_path_info.push(index_key)?; - // the index path is now something likeDataContracts/ContractID/Documents(1)/$ownerId/ + // The final bucket takes ownership of `index_path`; + // earlier buckets (only a time-range fan-out has + // more than one) clone it. + let own_index_path = if bucket + 1 == bucket_count { + std::mem::take(&mut index_path) + } else { + index_path.clone() + }; + let mut index_path_info = if document_and_contract_info + .owned_document_info + .document_info + .is_document_size() + { + // This is a stateless operation + PathInfo::PathWithSizes(KeyInfoPath::from_known_owned_path( + own_index_path, + )) + } else { + PathInfo::PathAsVec::<0>(own_index_path) + }; - let index_batch_operations: &mut Vec = - if sub_level_is_ephemeral { - &mut ephemeral_local_operations - } else { - &mut *batch_operations - }; - self.remove_indices_for_index_level_for_contract_operations( - document_and_contract_info, - index_path_info, - sub_level, - any_fields_null, - all_fields_null, - value_tree_type, - &index_storage_flags, - previous_batch_operations, - estimated_costs_only_with_layer_info, - skip_missing_expired_entry, - event_id, - transaction, - index_batch_operations, - platform_version, - )?; - } + // we push the actual value of the index path + index_path_info.push(index_key)?; + // the index path is now something likeDataContracts/ContractID/Documents(1)/$ownerId/ - if sub_level_is_ephemeral { - batch_operations.extend( - ephemeral_local_operations - .into_iter() - .map(LowLevelDriveOperation::retag_ephemeral), - ); - } + self.remove_indices_for_index_level_for_contract_operations( + document_and_contract_info, + index_path_info, + sub_level, + any_fields_null, + all_fields_null, + value_tree_type, + &index_storage_flags, + previous_batch_operations, + estimated_costs_only_with_layer_info, + skip_missing_expired_entry, + event_id, + transaction, + index_batch_operations, + platform_version, + )?; + } + Ok(()) + }, + )?; } Ok(()) } diff --git a/packages/rs-drive/src/drive/document/delete/remove_reference_for_index_level_for_contract_operations/v1/mod.rs b/packages/rs-drive/src/drive/document/delete/remove_reference_for_index_level_for_contract_operations/v1/mod.rs index 5f5b2ea2447..0a47f746590 100644 --- a/packages/rs-drive/src/drive/document/delete/remove_reference_for_index_level_for_contract_operations/v1/mod.rs +++ b/packages/rs-drive/src/drive/document/delete/remove_reference_for_index_level_for_contract_operations/v1/mod.rs @@ -102,12 +102,13 @@ impl Drive { path_segments.push(vec![0]); } // The document-type path plus the (grid-qualified) index level - // key exist for every registered contract: the flag is only - // ever set for a bucketed index, whose level tree is created at - // contract registration — so the walk starts below them. + // key exist for every registered contract, and the flag is only + // set once the top-level walker found the bucket standing + // (`time_range_entry_state`) — so the walk starts below the + // bucket. if !self.expired_entry_path_exists( &path_segments, - usize::from(CONTRACT_DOCUMENTS_PATH_HEIGHT) + 1, + usize::from(CONTRACT_DOCUMENTS_PATH_HEIGHT) + 2, transaction, platform_version, )? { diff --git a/packages/rs-drive/src/drive/document/insert/add_document_for_contract/time_range_index_e2e_tests.rs b/packages/rs-drive/src/drive/document/insert/add_document_for_contract/time_range_index_e2e_tests.rs index 77e4904c181..f976e6c6b05 100644 --- a/packages/rs-drive/src/drive/document/insert/add_document_for_contract/time_range_index_e2e_tests.rs +++ b/packages/rs-drive/src/drive/document/insert/add_document_for_contract/time_range_index_e2e_tests.rs @@ -2403,16 +2403,18 @@ fn ttl_partial_drain_resumes_across_writes_and_removals_stay_exact() { let t0 = 2_000 * h; let old_bucket_key = DocumentPropertyType::encode_date_timestamp(t0); - // Five groups in the doomed bucket: full drainage costs - // 5 × ([0] drop + value-tree delete) + property-name drop + bucket - // drop = 12 operations, above the per-write budget of 8. - let docs: Vec = (1..=5) - .map(|i| insert_at(t0 + i * MINUTE_MS_TTL, &format!("g{i}"))) + // Fourteen groups in the doomed bucket: full drainage costs + // 14 × ([0] drop + value-tree delete) + property-name drop + bucket + // drop = 30 operations — enough that the first write and the two + // deletes below (every write drains, deletes included) each spend a + // full 8-op budget without finishing it. + let docs: Vec = (1..=14) + .map(|i| insert_at(t0 + i * MINUTE_MS_TTL, &format!("g{i:02}"))) .collect(); - // First write past the horizon: budget 8 drains groups g1..g4 (2 ops - // each) and stops — the bucket stands, partially drained, with g5 and - // the property-name tree intact. + // First write past the horizon: budget 8 drains groups g01..g04 (2 ops + // each) and stops — the bucket stands, partially drained, with + // g05..g14 and the property-name tree intact. insert_at(t0 + 6 * h, "w1"); let bucket_path = { let mut path = level_path.clone(); @@ -2429,18 +2431,24 @@ fn ttl_partial_drain_resumes_across_writes_and_removals_stay_exact() { path.push(tag.as_bytes().to_vec()); path }; - for gone in ["g1", "g2", "g3", "g4"] { + for gone in ["g01", "g02", "g03", "g04"] { assert!( !path_exists(&group_path(gone)), "group {gone} drains in the first write" ); } - assert!(path_exists(&group_path("g5")), "the budget stops before g5"); + assert!( + path_exists(&group_path("g05")), + "the budget stops before g05" + ); + assert!(path_exists(&group_path("g14")), "g14 stands untouched"); - // A document whose group the drain took deletes as a clean skip; one - // whose group still stands deletes normally. Both under the standing, - // partially drained bucket. - for (doc, label) in [(&docs[0], "drained group"), (&docs[4], "standing group")] { + // A document whose group still stands deletes normally; one whose + // group the drain took deletes as a clean skip. Both under the + // standing, partially drained bucket — and each delete's own drain + // spends another budget (g06..g09, then g10..g13), so the standing + // group must go first. + for (doc, label) in [(&docs[4], "standing group"), (&docs[0], "drained group")] { drive .delete_document_for_contract( doc.id(), @@ -2458,8 +2466,23 @@ fn ttl_partial_drain_resumes_across_writes_and_removals_stay_exact() { .unwrap_or_else(|e| panic!("deleting a doc from a {label} must succeed: {e:?}")); } - // The next write finishes whatever drainage the deletes' own up-tree - // pruning left behind; the bucket is gone. + // The first delete's up-tree pruning takes the emptied g05 chain and + // its drain g06..g09; the second delete's drain takes g10..g13. The + // property-name tree still holds g14, so the bucket stands, and only + // the next write's drain can finish it. + assert!( + !path_exists(&group_path("g05")), + "deleting g05's only document prunes the group" + ); + assert!( + !path_exists(&group_path("g13")), + "the deletes' drains reach g13" + ); + assert!(path_exists(&group_path("g14")), "g14 still stands"); + assert!( + path_exists(&bucket_path), + "the bucket stands until drainage resumes" + ); insert_at(t0 + 6 * h + 20 * MINUTE_MS_TTL, "w2"); assert!( !path_exists(&bucket_path), @@ -2712,7 +2735,7 @@ fn ttl_drainage_covers_every_parent_layout() { #[test] fn ttl_budget_boundary_after_zero_tree_keeps_deletes_exact() { use crate::drive::document::paths::contract_document_type_path_vec; - use crate::fees::op::LowLevelDriveOperation; + use crate::fees::op::{LowLevelDriveOperation, TimeRangeTtlDrainRequest}; use crate::util::grove_operations::DirectQueryType; use dpp::data_contract::document_type::DocumentPropertyType; use grovedb_path::SubtreePath; @@ -2803,16 +2826,15 @@ fn ttl_budget_boundary_after_zero_tree_keeps_deletes_exact() { // Budget 1: exactly the group's `[0]` tree drops; its value tree // stands without it. + let drain_request = |max_operations: u16| TimeRangeTtlDrainRequest { + transform: transform.clone(), + bucket_level: bucket_level.clone(), + level_path: level_path.clone(), + block_time_ms: after_expiry_ms, + max_operations, + }; drive - .drain_expired_time_range_buckets( - &transform, - bucket_level, - &level_path, - after_expiry_ms, - 1, - None, - platform_version, - ) + .drain_expired_time_range_buckets(&drain_request(1), None, &platform_version.drive) .expect("a budget of one drops exactly the [0] tree"); let exists = |segments: &[Vec]| -> bool { let (key, parents) = segments.split_last().expect("non-empty"); @@ -2859,15 +2881,7 @@ fn ttl_budget_boundary_after_zero_tree_keeps_deletes_exact() { // A later drain finishes the bucket. drive - .drain_expired_time_range_buckets( - &transform, - bucket_level, - &level_path, - after_expiry_ms, - 16, - None, - platform_version, - ) + .drain_expired_time_range_buckets(&drain_request(16), None, &platform_version.drive) .expect("the rest of the bucket drains"); let mut bucket_path = level_path.clone(); bucket_path.push(bucket_key); @@ -3292,6 +3306,146 @@ fn ttl_index_bytes_bill_to_processing_without_refunds() { ); } +/// Regression: drainage must never run while a transition's own removals +/// are still queued. A documents batch `[delete D, create E]` where D's +/// window is expired but still standing queues removals under that +/// bucket; a drain executed inline by a LATER transition of the same +/// batch (before its own operations are queued, or worse, after) can +/// flat-drop the very subtrees those removals target, and the batch +/// apply fails with `InvalidPath`. Filler groups sort before D's group so +/// the delete's own budget runs out before reaching it — D's trees stand +/// when its removal is queued, and the create's budget would take them. +/// The drain now runs once per level after the batch has applied. +#[test] +fn ttl_removals_queued_before_a_draining_write_in_one_batch_still_apply() { + use crate::drive::document::paths::contract_document_type_path_vec; + use crate::fees::op::LowLevelDriveOperation; + use crate::util::batch::{DocumentOperationType, DriveOperation}; + use crate::util::grove_operations::DirectQueryType; + use crate::util::object_size_info::{DataContractInfo, DocumentTypeInfo}; + use dpp::data_contract::document_type::DocumentPropertyType; + use grovedb_path::SubtreePath; + + let platform_version = PlatformVersion::latest(); + let drive = setup_drive_with_initial_state_structure(Some(platform_version)); + let contract = build_ttl_contract_with_index_keys(218, vec![]); + drive + .apply_contract( + &contract, + BlockInfo::default(), + true, + StorageFlags::optional_default_as_cow(), + None, + platform_version, + ) + .expect("apply contract"); + let document_type = contract.document_type_for_name("post").expect("post"); + let transform = document_type + .indexes() + .get("trendingTtl") + .expect("index") + .time_range + .clone() + .expect("transform"); + + let h = HOUR_MS; + let t0 = 8_000 * h; + let make_doc = |marker: u8, created_at: u64, tag: &str| -> Document { + Document::V0(DocumentV0 { + id: Identifier::from(fixture_bytes(marker, created_at, tag)), + owner_id: Identifier::from(fixture_bytes(marker.wrapping_add(1), created_at, tag)), + properties: BTreeMap::from([ + ("hashtag".to_string(), Value::Text(tag.to_string())), + ("amount".to_string(), Value::U64(3)), + ]), + created_at: Some(created_at), + revision: Some(1), + ..Default::default() + }) + }; + fn add_op<'a>(contract: &'a DataContract, document: &'a Document) -> DriveOperation<'a> { + DriveOperation::DocumentOperation(DocumentOperationType::AddDocument { + owned_document_info: OwnedDocumentInfo { + document_info: DocumentRefInfo((document, None)), + owner_id: Some(document.owner_id().to_buffer()), + }, + contract_info: DataContractInfo::BorrowedDataContract(contract), + document_type_info: DocumentTypeInfo::DocumentTypeNameAsStr("post"), + override_document: false, + }) + } + let apply = |operations: Vec, time_ms: u64| { + drive.apply_drive_operations( + operations, + true, + &BlockInfo { + time_ms, + ..Default::default() + }, + None, + platform_version, + None, + ) + }; + + // Five filler groups (a1..a5) sort before D's group (zz): the doomed + // bucket costs 6 × 2 + 1 + 1 = 14 drop operations, so one 8-op budget + // takes a1..a4 and leaves zz standing. No write happens between the + // inserts and expiry. + let fillers: Vec = (1..=5) + .map(|i| make_doc(30 + 2 * i, t0 + MINUTE_MS_TTL, &format!("a{i}"))) + .collect(); + let d = make_doc(20, t0 + MINUTE_MS_TTL, "zz"); + for document in fillers.iter().chain(std::iter::once(&d)) { + apply(vec![add_op(&contract, document)], t0 + MINUTE_MS_TTL).expect("insert"); + } + + let mut level_path = contract_document_type_path_vec(contract.id_ref().as_bytes(), "post"); + level_path.push(transform.storage_key("$createdAt").into_bytes()); + let exists = |segments: &[Vec]| -> bool { + let (key, parents) = segments.split_last().expect("non-empty"); + let mut ops: Vec = vec![]; + drive + .grove_has_raw( + SubtreePath::from(parents), + key.as_slice(), + DirectQueryType::StatefulDirectQuery, + None, + &mut ops, + &platform_version.drive, + ) + .expect("existence check") + }; + let mut old_bucket_path = level_path.clone(); + old_bucket_path.push(DocumentPropertyType::encode_date_timestamp(t0)); + assert!(exists(&old_bucket_path), "the bucket stands before expiry"); + + // Past the horizon: delete D (queued under the expired-but-standing + // bucket, in the group the delete's own budget does not reach), then + // create E in the same batch. + let e = make_doc(24, t0 + 6 * h, "e"); + apply( + vec![ + DriveOperation::DocumentOperation(DocumentOperationType::DeleteDocument { + document_id: d.id(), + contract_info: DataContractInfo::BorrowedDataContract(&contract), + document_type_info: DocumentTypeInfo::DocumentTypeNameAsStr("post"), + }), + add_op(&contract, &e), + ], + t0 + 6 * h, + ) + .expect("removals queued before a draining write must still apply"); + + let mut zz_path = old_bucket_path.clone(); + zz_path.push(b"hashtag".to_vec()); + zz_path.push(b"zz".to_vec()); + assert!(!exists(&zz_path), "D's group is gone with D"); + let mut new_bucket_path = level_path.clone(); + new_bucket_path.push(DocumentPropertyType::encode_date_timestamp(t0 + 6 * h)); + assert!(exists(&new_bucket_path), "the live bucket took the create"); +} + /// Several indexes may share one grid-qualified level (same grid, same /// ttl); the walkers must drain that level exactly ONCE per write, before /// any batch mutation is queued. The per-index regression: four countable @@ -3444,10 +3598,53 @@ fn ttl_shared_grid_drains_once_per_write() { None, ) .expect("add document"); + // A second document in the same bucket, distinct on every indexed + // property: the update below prunes its own emptied chains on all + // four indexes, so without a survivor the bucket would empty and go + // through up-tree pruning rather than through the drain. + let survivor_owner = fixture_bytes(21, t0, "eta"); + let survivor = Document::V0(DocumentV0 { + id: Identifier::from(fixture_bytes(22, t0, "eta")), + owner_id: Identifier::from(survivor_owner), + properties: BTreeMap::from([ + ("hashtag".to_string(), Value::Text("eta".to_string())), + ("amount".to_string(), Value::U64(7)), + ("alpha".to_string(), Value::Text("seven".to_string())), + ("beta".to_string(), Value::Text("ten".to_string())), + ]), + created_at: Some(t0 + MINUTE_MS_TTL), + revision: Some(1), + ..Default::default() + }); + drive + .add_document_for_contract( + DocumentAndContractInfo { + owned_document_info: OwnedDocumentInfo { + document_info: DocumentRefInfo(( + &survivor, + StorageFlags::optional_default_as_cow(), + )), + owner_id: Some(survivor_owner), + }, + contract: &contract, + document_type, + }, + false, + BlockInfo { + time_ms: t0 + MINUTE_MS_TTL, + ..Default::default() + }, + true, + None, + platform_version, + None, + ) + .expect("add survivor"); - // First write past the horizon: the bucket needs 13 drop operations, - // the budget allows 8 — a partial drain is guaranteed, and every - // index's queued removals must stay consistent with it. + // First write past the horizon: the survivor's entries leave the + // bucket needing 13 drop operations, the budget allows 8 — a partial + // drain is guaranteed, and every index's queued removals must stay + // consistent with it. let mut update_at = |time_ms: u64, revision: u64, hashtag: &str| { document.set("hashtag", Value::Text(hashtag.to_string())); document.set_revision(Some(revision)); diff --git a/packages/rs-drive/src/drive/document/insert/add_indices_for_top_index_level_for_contract_operations/v2/mod.rs b/packages/rs-drive/src/drive/document/insert/add_indices_for_top_index_level_for_contract_operations/v2/mod.rs index a3e1291ce87..159a7f0dd15 100644 --- a/packages/rs-drive/src/drive/document/insert/add_indices_for_top_index_level_for_contract_operations/v2/mod.rs +++ b/packages/rs-drive/src/drive/document/insert/add_indices_for_top_index_level_for_contract_operations/v2/mod.rs @@ -8,9 +8,10 @@ use crate::util::object_size_info::{ DocumentAndContractInfo, DocumentInfoV0Methods, DriveKeyInfo, PathInfo, }; +use crate::drive::document::time_range_ttl::live_time_range_index_keys; use crate::error::fee::FeeError; use crate::error::Error; -use crate::fees::op::LowLevelDriveOperation; +use crate::fees::op::{LowLevelDriveOperation, TimeRangeTtlDrainRequest}; use dpp::data_contract::accessors::v0::DataContractV0Getters; use dpp::data_contract::config::v0::DataContractConfigGettersV0; use dpp::data_contract::document_type::accessors::{DocumentTypeV0Getters, DocumentTypeV2Getters}; @@ -132,7 +133,6 @@ impl Drive { let sub_level_is_ephemeral = sub_level .time_range() .is_some_and(|transform| transform.ttl_seconds.is_some()); - let mut ephemeral_local_operations: Vec = vec![]; let index_storage_flags = if sub_level_is_ephemeral { None } else { @@ -201,7 +201,7 @@ impl Drive { BatchInsertTreeApplyType::StatelessBatchInsertTree { in_tree_type: property_name_tree_type, tree_type: value_tree_type, - flags_len: storage_flags + flags_len: index_storage_flags .map(|s| s.serialized_size()) .unwrap_or_default(), } @@ -267,12 +267,26 @@ impl Drive { .unwrap_or(1), ); + // TTL: writes never target expired buckets. Consensus assigns + // the source timestamp from block time, so with `ttl >= range` + // only a re-inserted document with a historical timestamp (a + // contested document awarded after its window expired) loses + // keys here — it simply gets no entries under this index, and + // never re-creates a dropped path. Size-only keys pass. + let index_keys = match sub_level.time_range() { + Some(transform) if transform.ttl_seconds.is_some() => { + live_time_range_index_keys(transform, index_keys, block_time_ms) + } + _ => index_keys, + }; + // TTL drainage rides every write into a TTL'd index: a bounded // number of deepest-first drop operations against the oldest // expired bucket, resuming wherever the previous write's budget - // ran out. When nothing is expired this is one bounded range - // read. Stateful only — the estimation dry run neither reads - // state nor prices drops (each is O(1); the count is capped). + // ran out. The drain is requested here and run once the + // transition's batch has applied (one drain per level — see + // `apply_batch_low_level_drive_operations`). Stateful only — + // the estimation dry run neither reads state nor prices drops. if estimated_costs_only_with_layer_info.is_none() { if let Some(transform) = sub_level.time_range() { if transform.ttl_seconds.is_some() { @@ -280,98 +294,87 @@ impl Drive { .system_limits .max_time_range_ttl_drop_operations_per_write { - self.drain_expired_time_range_buckets( - transform, - sub_level, - &index_path, - block_time_ms, - max_operations, - transaction, - platform_version, - )?; + batch_operations.push(LowLevelDriveOperation::TimeRangeTtlDrain( + TimeRangeTtlDrainRequest { + transform: transform.clone(), + bucket_level: sub_level.clone(), + level_path: index_path.clone(), + block_time_ms, + max_operations, + }, + )); } } } } let bucket_count = index_keys.len(); - for (bucket, index_key) in index_keys.into_iter().enumerate() { - // The zero will not matter here, because the PathKeyInfo is variable - let path_key_info = index_key.clone().add_path::<0>(index_path.clone()); - let index_batch_operations: &mut Vec = - if sub_level_is_ephemeral { - &mut ephemeral_local_operations - } else { - &mut *batch_operations - }; - self.batch_insert_empty_tree_if_not_exists( - path_key_info, - value_tree_type, - index_storage_flags, - value_apply_type, - transaction, - previous_batch_operations, - index_batch_operations, - drive_version, - )?; - - // The final bucket takes ownership of `index_path`; earlier - // buckets (only a time-range fan-out has more than one) - // clone it. - let own_index_path = if bucket + 1 == bucket_count { - std::mem::take(&mut index_path) - } else { - index_path.clone() - }; - let mut index_path_info = if document_and_contract_info - .owned_document_info - .document_info - .is_document_size() - { - // This is a stateless operation - PathInfo::PathWithSizes(KeyInfoPath::from_known_owned_path(own_index_path)) - } else { - PathInfo::PathAsVec::<0>(own_index_path) - }; + LowLevelDriveOperation::with_ephemeral_routing( + batch_operations, + sub_level_is_ephemeral, + |index_batch_operations| { + for (bucket, index_key) in index_keys.into_iter().enumerate() { + // The zero will not matter here, because the PathKeyInfo is variable + let path_key_info = index_key.clone().add_path::<0>(index_path.clone()); + self.batch_insert_empty_tree_if_not_exists( + path_key_info, + value_tree_type, + index_storage_flags, + value_apply_type, + transaction, + previous_batch_operations, + index_batch_operations, + drive_version, + )?; - // we push the actual value of the index path - index_path_info.push(index_key)?; - // the index path is now something likeDataContracts/ContractID/Documents(1)/$ownerId/ + // The final bucket takes ownership of `index_path`; + // earlier buckets (only a time-range fan-out has + // more than one) clone it. + let own_index_path = if bucket + 1 == bucket_count { + std::mem::take(&mut index_path) + } else { + index_path.clone() + }; + let mut index_path_info = if document_and_contract_info + .owned_document_info + .document_info + .is_document_size() + { + // This is a stateless operation + PathInfo::PathWithSizes(KeyInfoPath::from_known_owned_path( + own_index_path, + )) + } else { + PathInfo::PathAsVec::<0>(own_index_path) + }; - // Propagate the exact (post-demotion) `value_tree_type` we - // just inserted forward as the recursive level's - // `parent_value_tree_type` so its continuation children pick - // the right zero-contribution op. - let index_batch_operations: &mut Vec = - if sub_level_is_ephemeral { - &mut ephemeral_local_operations - } else { - &mut *batch_operations - }; - self.add_indices_for_index_level_for_contract_operations( - document_and_contract_info, - index_path_info, - sub_level, - any_fields_null, - all_fields_null, - value_tree_type, - previous_batch_operations, - &index_storage_flags, - estimated_costs_only_with_layer_info, - event_id, - transaction, - index_batch_operations, - platform_version, - )?; - } + // we push the actual value of the index path + index_path_info.push(index_key)?; + // the index path is now something likeDataContracts/ContractID/Documents(1)/$ownerId/ - if sub_level_is_ephemeral { - batch_operations.extend( - ephemeral_local_operations - .into_iter() - .map(LowLevelDriveOperation::retag_ephemeral), - ); - } + // Propagate the exact (post-demotion) `value_tree_type` + // we just inserted forward as the recursive level's + // `parent_value_tree_type` so its continuation + // children pick the right zero-contribution op. + self.add_indices_for_index_level_for_contract_operations( + document_and_contract_info, + index_path_info, + sub_level, + any_fields_null, + all_fields_null, + value_tree_type, + previous_batch_operations, + &index_storage_flags, + estimated_costs_only_with_layer_info, + event_id, + transaction, + index_batch_operations, + platform_version, + )?; + } + Ok(()) + }, + )?; } Ok(()) } diff --git a/packages/rs-drive/src/drive/document/insert/add_reference_for_index_level_for_contract_operations/mod.rs b/packages/rs-drive/src/drive/document/insert/add_reference_for_index_level_for_contract_operations/mod.rs index 08bedec07d7..f91c392e7b7 100644 --- a/packages/rs-drive/src/drive/document/insert/add_reference_for_index_level_for_contract_operations/mod.rs +++ b/packages/rs-drive/src/drive/document/insert/add_reference_for_index_level_for_contract_operations/mod.rs @@ -81,7 +81,7 @@ impl Drive { ), version => Err(Error::Drive(DriveError::UnknownVersionMismatch { method: "add_reference_for_index_level_for_contract_operations".to_string(), - known_versions: vec![0], + known_versions: vec![0, 1], received: version, })), } diff --git a/packages/rs-drive/src/drive/document/insert/add_reference_for_index_level_for_contract_operations/v0/mod.rs b/packages/rs-drive/src/drive/document/insert/add_reference_for_index_level_for_contract_operations/v0/mod.rs index 9ecc71a405d..9c9252e1143 100644 --- a/packages/rs-drive/src/drive/document/insert/add_reference_for_index_level_for_contract_operations/v0/mod.rs +++ b/packages/rs-drive/src/drive/document/insert/add_reference_for_index_level_for_contract_operations/v0/mod.rs @@ -34,11 +34,68 @@ use grovedb::EstimatedLayerSizes::{AllItems, AllReference}; use grovedb::{Element, EstimatedLayerInformation, TransactionArg, TreeType}; use std::collections::HashMap; +/// Which `Option<&StorageFlags>` the terminal reference element is built +/// with: v0 reads the document info's own flags, v1 the walker-passed ones. +#[derive(Clone, Copy)] +pub(super) enum TerminalReferenceFlagsSource { + DocumentInfo, + Walker, +} + +impl TerminalReferenceFlagsSource { + fn select<'a>( + self, + document_info: Option<&'a StorageFlags>, + walker: Option<&'a StorageFlags>, + ) -> Option<&'a StorageFlags> { + match self { + Self::DocumentInfo => document_info, + Self::Walker => walker, + } + } +} + impl Drive { /// Adds the terminal reference. #[inline(always)] #[allow(clippy::too_many_arguments)] pub(super) fn add_reference_for_index_level_for_contract_operations_v0( + &self, + document_and_contract_info: &DocumentAndContractInfo, + index_path_info: PathInfo<0>, + // See the wrapper's docstring for why this is a borrow now. + index_type: &IndexLevelTypeInfo, + any_fields_null: bool, + all_fields_null: bool, + previous_batch_operations: &mut Option<&mut Vec>, + storage_flags: &Option<&StorageFlags>, + estimated_costs_only_with_layer_info: &mut Option< + HashMap, + >, + transaction: TransactionArg, + batch_operations: &mut Vec, + platform_version: &PlatformVersion, + ) -> Result<(), Error> { + self.add_reference_for_index_level_for_contract_operations_inner( + document_and_contract_info, + index_path_info, + index_type, + any_fields_null, + all_fields_null, + previous_batch_operations, + storage_flags, + estimated_costs_only_with_layer_info, + transaction, + batch_operations, + platform_version, + TerminalReferenceFlagsSource::DocumentInfo, + ) + } + + /// Shared body of the v0 and v1 terminal-reference insert; the versions + /// differ only in `terminal_flags_source`. + #[allow(clippy::too_many_arguments)] + pub(super) fn add_reference_for_index_level_for_contract_operations_inner( &self, document_and_contract_info: &DocumentAndContractInfo, mut index_path_info: PathInfo<0>, @@ -54,6 +111,7 @@ impl Drive { transaction: TransactionArg, batch_operations: &mut Vec, platform_version: &PlatformVersion, + terminal_flags_source: TerminalReferenceFlagsSource, ) -> Result<(), Error> { let drive_version = &platform_version.drive; @@ -205,19 +263,25 @@ impl Drive { let key_element_info = match &document_and_contract_info.owned_document_info.document_info { - DocumentRefAndSerialization((document, _, storage_flags)) - | DocumentRefInfo((document, storage_flags)) => { + DocumentRefAndSerialization((document, _, document_flags)) + | DocumentRefInfo((document, document_flags)) => { let document_reference = make_terminal_ref( document, - storage_flags.as_ref().map(|flags| flags.as_ref()), + terminal_flags_source.select( + document_flags.as_ref().map(|flags| flags.as_ref()), + *storage_flags, + ), )?; KeyElement((document.id_ref().as_slice(), document_reference)) } - DocumentOwnedInfo((document, storage_flags)) - | DocumentAndSerialization((document, _, storage_flags)) => { + DocumentOwnedInfo((document, document_flags)) + | DocumentAndSerialization((document, _, document_flags)) => { let document_reference = make_terminal_ref( document, - storage_flags.as_ref().map(|flags| flags.as_ref()), + terminal_flags_source.select( + document_flags.as_ref().map(|flags| flags.as_ref()), + *storage_flags, + ), )?; KeyElement((document.id_ref().as_slice(), document_reference)) } @@ -263,19 +327,25 @@ impl Drive { } else { let key_element_info = match &document_and_contract_info.owned_document_info.document_info { - DocumentRefAndSerialization((document, _, storage_flags)) - | DocumentRefInfo((document, storage_flags)) => { + DocumentRefAndSerialization((document, _, document_flags)) + | DocumentRefInfo((document, document_flags)) => { let document_reference = make_terminal_ref( document, - storage_flags.as_ref().map(|flags| flags.as_ref()), + terminal_flags_source.select( + document_flags.as_ref().map(|flags| flags.as_ref()), + *storage_flags, + ), )?; KeyElement((&[0], document_reference)) } - DocumentOwnedInfo((document, storage_flags)) - | DocumentAndSerialization((document, _, storage_flags)) => { + DocumentOwnedInfo((document, document_flags)) + | DocumentAndSerialization((document, _, document_flags)) => { let document_reference = make_terminal_ref( document, - storage_flags.as_ref().map(|flags| flags.as_ref()), + terminal_flags_source.select( + document_flags.as_ref().map(|flags| flags.as_ref()), + *storage_flags, + ), )?; KeyElement((&[0], document_reference)) } diff --git a/packages/rs-drive/src/drive/document/insert/add_reference_for_index_level_for_contract_operations/v1/mod.rs b/packages/rs-drive/src/drive/document/insert/add_reference_for_index_level_for_contract_operations/v1/mod.rs index 1d0e6a20257..82960651b73 100644 --- a/packages/rs-drive/src/drive/document/insert/add_reference_for_index_level_for_contract_operations/v1/mod.rs +++ b/packages/rs-drive/src/drive/document/insert/add_reference_for_index_level_for_contract_operations/v1/mod.rs @@ -1,34 +1,13 @@ -use crate::drive::constants::STORAGE_FLAGS_SIZE; -use crate::drive::document::index_level_tree_types::terminal_member_tree_type; -use crate::drive::document::{ - document_reference_size, make_document_reference, make_document_reference_with_sum_item, - read_document_sum_contribution, -}; +use super::v0::TerminalReferenceFlagsSource; use crate::drive::Drive; -use crate::error::drive::DriveError; use crate::error::Error; use crate::fees::op::LowLevelDriveOperation; -use crate::util::grove_operations::QueryTarget::QueryTargetValue; -use crate::util::grove_operations::{BatchInsertApplyType, BatchInsertTreeApplyType}; -use crate::util::object_size_info::DocumentInfo::{ - DocumentAndSerialization, DocumentEstimatedAverageSize, DocumentOwnedInfo, - DocumentRefAndSerialization, DocumentRefInfo, -}; -use crate::util::object_size_info::DriveKeyInfo::{Key, KeyRef}; -use crate::util::object_size_info::KeyElementInfo::{KeyElement, KeyUnknownElementSize}; -use crate::util::object_size_info::{DocumentAndContractInfo, PathInfo, PathKeyElementInfo}; +use crate::util::object_size_info::{DocumentAndContractInfo, PathInfo}; use crate::util::storage_flags::StorageFlags; -use crate::util::type_constants::DEFAULT_HASH_SIZE_U8; -use dpp::data_contract::document_type::methods::DocumentTypeBasicMethods; use dpp::data_contract::document_type::IndexLevelTypeInfo; -use dpp::document::Document; -use dpp::document::DocumentV0Getters; use dpp::version::PlatformVersion; -use grovedb::batch::key_info::KeyInfo; use grovedb::batch::KeyInfoPath; -use grovedb::EstimatedLayerCount::PotentiallyAtMaxElements; -use grovedb::EstimatedLayerSizes::AllReference; -use grovedb::{Element, EstimatedLayerInformation, TransactionArg, TreeType}; +use grovedb::{EstimatedLayerInformation, TransactionArg}; use std::collections::HashMap; impl Drive { @@ -45,8 +24,7 @@ impl Drive { pub(super) fn add_reference_for_index_level_for_contract_operations_v1( &self, document_and_contract_info: &DocumentAndContractInfo, - mut index_path_info: PathInfo<0>, - // See the wrapper's docstring for why this is a borrow now. + index_path_info: PathInfo<0>, index_type: &IndexLevelTypeInfo, any_fields_null: bool, all_fields_null: bool, @@ -59,282 +37,19 @@ impl Drive { batch_operations: &mut Vec, platform_version: &PlatformVersion, ) -> Result<(), Error> { - let drive_version = &platform_version.drive; - - if all_fields_null && !index_type.should_insert_with_all_null { - return Ok(()); - } - - // indexOnly terminal: the member key is the terminal property's - // value and the element is an empty `Item` — there is no - // primary-storage row to reference. `terminal` can only be `Some` - // on a PV14+ indexOnly contract (the grammar rejects the keyword - // below meta-schema v3), so this branch is unreachable for every - // historical document — the same in-place gating the count and sum - // flags in this function already rely on. - if let Some(terminal_property) = index_type.terminal.as_deref() { - return self.add_index_only_terminal_item_operations( - document_and_contract_info, - index_path_info, - index_type, - terminal_property, - previous_batch_operations, - storage_flags, - estimated_costs_only_with_layer_info, - transaction, - batch_operations, - platform_version, - ); - } - - // The terminal reference's tree type is driven by the - // composition of the index's countability AND summability, - // per-axis (grovedb PR 670's expanded TreeType set - // distinguishes provable from root-only on each axis - // independently): - // - // - count provable + sum root → `ProvableCountSumTree` - // (existing variant: per-node count, root-only sum) - // - count root + sum provable → `ProvableCountProvableSumTree` - // (no dedicated "count-root + sum-provable" variant exists; - // upgrades count to per-node too) - // - count provable + sum provable → - // `ProvableCountProvableSumTree` (PR 670 newcomer: both - // per-node) - // - // Same dispatch shape as the primary-key tree dispatcher's v1 - // arm in `primary_key_tree_type.rs` — see - // `terminal_member_tree_type` for the full table (shared with - // the delete side and the indexOnly terminal branches). The - // `IndexLevelTypeInfo`'s `summable` carries the property name - // the reference's sum-item will contribute (read below to - // construct the `Element::ReferenceWithSumItem` that replaces a - // plain `Element::Reference` under summable indexes). - let reference_tree_type = terminal_member_tree_type(index_type); - - // Element-shape selector. Under a summable index path the - // reference element MUST be - // `Element::ReferenceWithSumItem(reference_path, amount_i64, - // flags)` (grovedb PR 670) rather than a plain - // `Element::Reference` — only `ReferenceWithSumItem` - // contributes a sum to the ancestor sum trees while still - // dereferencing to the document body in primary storage - // (so document iteration via index walks keeps working - // identically to the count side). Read the sum contribution - // once per insert from the document's `summable.unwrap()` - // property and freeze it into the element. On delete, grovedb - // pulls the same sum value off the stored element and - // propagates the subtraction up the merk path — no need to - // re-read the source document on the way down. - let sum_property_name: Option<&str> = index_type.summable.as_deref(); - let make_terminal_ref = - |document: &Document, storage_flags: Option<&StorageFlags>| -> Result { - match sum_property_name { - Some(prop_name) => { - // DPP validator guarantees the property is in - // `required` and is an integer type, so this - // conversion is safe — propagated as - // `CorruptedCodeExecution` if it ever fails. - let sum_value = read_document_sum_contribution(document, prop_name)?; - Ok(make_document_reference_with_sum_item( - document, - document_and_contract_info.document_type, - sum_value, - storage_flags, - )) - } - None => Ok(make_document_reference( - document, - document_and_contract_info.document_type, - storage_flags, - )), - } - }; - // unique indexes will be stored under key "0" - // non-unique indices should have a tree at key "0" that has all elements based off of primary key - if !index_type.index_type.is_unique() || any_fields_null { - // Tree generation, this happens for both non unique indexes, unique indexes with a null inside - // a member of the path - let key_path_info = KeyRef(&[0]); - - let path_key_info = key_path_info.add_path_info(index_path_info.clone()); - - let apply_type = if estimated_costs_only_with_layer_info.is_none() { - BatchInsertTreeApplyType::StatefulBatchInsertTree - } else { - BatchInsertTreeApplyType::StatelessBatchInsertTree { - in_tree_type: TreeType::NormalTree, - tree_type: reference_tree_type, - flags_len: storage_flags - .map(|s| s.serialized_size()) - .unwrap_or_default(), - } - }; - - // Here we are inserting an empty tree that will have a subtree of all other index properties - // It is basically the 0 - // Underneath we will have all elements if non unique index, or all identity contenders if - // a contested resource index - self.batch_insert_empty_tree_if_not_exists( - path_key_info, - reference_tree_type, - *storage_flags, - apply_type, - transaction, - previous_batch_operations, - batch_operations, - drive_version, - )?; - - index_path_info.push(Key(vec![0]))?; - // This is the simpler situation - // Under each tree we have all the references - - if let Some(estimated_costs_only_with_layer_info) = estimated_costs_only_with_layer_info - { - // On this level we will have a 0 and all the top index paths - estimated_costs_only_with_layer_info.insert( - index_path_info.clone().convert_to_key_info_path(), - EstimatedLayerInformation { - tree_type: reference_tree_type, - estimated_layer_count: PotentiallyAtMaxElements, - estimated_layer_sizes: AllReference( - DEFAULT_HASH_SIZE_U8, - document_reference_size(document_and_contract_info.document_type), - storage_flags.map(|s| s.serialized_size()), - ), - }, - ); - } - - let key_element_info = match &document_and_contract_info - .owned_document_info - .document_info - { - DocumentRefAndSerialization((document, _, _)) | DocumentRefInfo((document, _)) => { - let document_reference = make_terminal_ref(document, *storage_flags)?; - KeyElement((document.id_ref().as_slice(), document_reference)) - } - DocumentOwnedInfo((document, _)) | DocumentAndSerialization((document, _, _)) => { - let document_reference = make_terminal_ref(document, *storage_flags)?; - KeyElement((document.id_ref().as_slice(), document_reference)) - } - DocumentEstimatedAverageSize(max_size) => KeyUnknownElementSize(( - KeyInfo::MaxKeySize { - unique_id: document_and_contract_info - .document_type - .unique_id_for_storage() - .to_vec(), - max_size: DEFAULT_HASH_SIZE_U8, - }, - // Match the sum-bearing variant the live path - // would have written: `make_document_reference_with_sum_item` - // emits `Element::ReferenceWithSumItem` when - // `sum_property_name.is_some()`. The sum-aware helper - // reserves 10 worst-case bytes for the i64 sum_value. - // Unconditional switch: this entire flow is v12+ - // gated (no v11 consensus baseline for sum-bearing - // index refs). - if sum_property_name.is_some() { - Element::required_reference_with_sum_item_space( - *max_size, - STORAGE_FLAGS_SIZE, - &drive_version.grove_version, - )? - } else { - Element::required_item_space( - *max_size, - STORAGE_FLAGS_SIZE, - &drive_version.grove_version, - )? - }, - )), - }; - - let path_key_element_info = PathKeyElementInfo::from_path_info_and_key_element( - index_path_info, - key_element_info, - )?; - - // here we should return an error if the element already exists - self.batch_insert(path_key_element_info, batch_operations, drive_version)?; - } else { - let key_element_info = match &document_and_contract_info - .owned_document_info - .document_info - { - DocumentRefAndSerialization((document, _, _)) | DocumentRefInfo((document, _)) => { - let document_reference = make_terminal_ref(document, *storage_flags)?; - KeyElement((&[0], document_reference)) - } - DocumentOwnedInfo((document, _)) | DocumentAndSerialization((document, _, _)) => { - let document_reference = make_terminal_ref(document, *storage_flags)?; - KeyElement((&[0], document_reference)) - } - DocumentEstimatedAverageSize(estimated_size) => KeyUnknownElementSize(( - KeyInfo::MaxKeySize { - unique_id: document_and_contract_info - .document_type - .unique_id_for_storage() - .to_vec(), - max_size: 1, - }, - // Parallel to the non-unique branch above: unique - // indexes with `summable: Some(_)` still write a - // `ReferenceWithSumItem` at the terminal `[0]` slot - // when there's any non-null entry (the unique-no-op - // caveat applies only to all-non-null exact matches, - // see book/document-sum-trees.md). The estimated - // worst-case treats the sum-bearing variant. - if sum_property_name.is_some() { - Element::required_reference_with_sum_item_space( - *estimated_size, - STORAGE_FLAGS_SIZE, - &drive_version.grove_version, - )? - } else { - Element::required_item_space( - *estimated_size, - STORAGE_FLAGS_SIZE, - &drive_version.grove_version, - )? - }, - )), - }; - - let path_key_element_info = PathKeyElementInfo::from_path_info_and_key_element( - index_path_info, - key_element_info, - )?; - - let apply_type = if estimated_costs_only_with_layer_info.is_none() { - BatchInsertApplyType::StatefulBatchInsert - } else { - BatchInsertApplyType::StatelessBatchInsert { - in_tree_type: reference_tree_type, - target: QueryTargetValue( - document_reference_size(document_and_contract_info.document_type) - + storage_flags - .map(|s| s.serialized_size()) - .unwrap_or_default(), - ), - } - }; - - // here we should return an error if the element already exists - let inserted = self.batch_insert_if_not_exists( - path_key_element_info, - apply_type, - transaction, - batch_operations, - drive_version, - )?; - if !inserted { - return Err(Error::Drive(DriveError::CorruptedContractIndexes( - "reference already exists".to_string(), - ))); - } - } - Ok(()) + self.add_reference_for_index_level_for_contract_operations_inner( + document_and_contract_info, + index_path_info, + index_type, + any_fields_null, + all_fields_null, + previous_batch_operations, + storage_flags, + estimated_costs_only_with_layer_info, + transaction, + batch_operations, + platform_version, + TerminalReferenceFlagsSource::Walker, + ) } } diff --git a/packages/rs-drive/src/drive/document/time_range_ttl.rs b/packages/rs-drive/src/drive/document/time_range_ttl.rs index 097b3c39f05..40a1af5f3d3 100644 --- a/packages/rs-drive/src/drive/document/time_range_ttl.rs +++ b/packages/rs-drive/src/drive/document/time_range_ttl.rs @@ -3,25 +3,36 @@ //! //! Three rules, one definition each: //! -//! - **Writes never target expired buckets.** The insert path cannot -//! produce one by construction (`$createdAt` &co. are consensus-assigned -//! and validation requires `ttl >= range`), and the update path filters -//! its new entry keys through [`live_time_range_entry_keys`] — an update -//! of a document whose windows have all expired simply leaves it with no -//! entries under the TTL'd index, and never resurrects a dropped bucket. +//! - **Writes never target expired buckets.** The insert walker filters +//! its bucket keys through [`live_time_range_index_keys`] and the update +//! walker its new entry keys through [`live_time_range_entry_keys`]: a +//! document whose windows have all expired simply gets no entries under +//! the TTL'd index, and never resurrects a dropped bucket. Consensus +//! assigns `$createdAt` &co. from block time, so with `ttl >= range` +//! only a re-inserted document with a historical timestamp (a contested +//! document awarded after its window expired) ever hits the filter. //! - **Removals touch an expired bucket only while it still stands.** The //! TTL drop is bucket-granular and lazy, so between a bucket's expiry //! and its drop a delete (or key-changing update) of one of its //! documents must still remove that document's entries — otherwise the //! bucket would carry dangling references until the drop. Once the //! bucket is gone the entries are gone with it, and per-entry removal -//! must skip rather than fail. [`Drive::time_range_entry_is_removable`] -//! is that check: live bucket ⇒ always removable; expired bucket ⇒ -//! removable exactly when it still exists. The existence read is -//! deterministic — it reads consensus state. +//! must skip rather than fail. [`Drive::time_range_entry_state`] is that +//! check: live bucket ⇒ always removable; expired bucket ⇒ removable +//! exactly when it still exists. The existence read is deterministic — +//! it reads consensus state. //! - **Expiry has one definition**: -//! [`TimeRangeTransform::expiry_horizon_ms`], shared by these helpers -//! and the bucket-drop cleanup. +//! [`TimeRangeTransform::bucket_expired`], shared by these helpers and +//! the bucket-drop cleanup. +//! +//! # Timing +//! +//! Drainage never runs while a transition's operations are still queued: +//! the walkers emit a [`TimeRangeTtlDrainRequest`] per write into a TTL'd +//! level, and `apply_batch_low_level_drive_operations` runs the requests — +//! once per level — after the batch is applied. Draining mid-conversion +//! could remove subtrees that queued removals (an earlier document of the +//! same transition, or an earlier index sharing the level) still targeted. //! //! # Billing //! @@ -32,19 +43,27 @@ //! invariant — the estimation dry run cannot read state, so it cannot //! price state-dependent drainage, and billing it on execution only would //! let a transition pass validation and then overdraw on apply. The work -//! itself is bounded (a capped count of O(1) operations plus a handful of -//! bounded reads per write) and is system maintenance of state nobody -//! holds refunds against; the planned ephemeral-bytes fee rate is where -//! TTL writers pre-pay it in aggregate. +//! itself is bounded (a capped count of drop operations applied as one +//! batch, plus a handful of bounded reads per write) and is system +//! maintenance of state nobody holds refunds against; the ephemeral-bytes +//! fee rate is where TTL writers pre-pay it in aggregate. -use crate::drive::document::index_level_tree_types::index_level_tree_types_with_continuation_demotion; +use crate::drive::document::index_level_tree_types::{ + index_level_tree_types_with_continuation_demotion, terminal_member_tree_type, +}; use crate::drive::Drive; +use crate::error::drive::DriveError; use crate::error::Error; -use crate::fees::op::LowLevelDriveOperation; +use crate::fees::op::{LowLevelDriveOperation, TimeRangeTtlDrainRequest}; +use crate::util::batch::grovedb_op_batch::GroveDbOpBatchV0Methods; +use crate::util::batch::GroveDbOpBatch; use crate::util::grove_operations::push_drive_operation_result; use crate::util::grove_operations::DirectQueryType; +use crate::util::object_size_info::DriveKeyInfo; use dpp::data_contract::document_type::{DocumentPropertyType, IndexLevel, TimeRangeTransform}; +use dpp::version::drive_versions::DriveVersion; use dpp::version::PlatformVersion; +use grovedb::batch::{QualifiedGroveDbOp, SubelementsDeletionBehavior}; use grovedb::query_result_type::QueryResultType; use grovedb::{PathQuery, Query, SizedQuery, TransactionArg, TreeType}; use grovedb_path::SubtreePath; @@ -60,6 +79,13 @@ pub(crate) fn entry_key_bucket_start(entry_key: &[u8]) -> Option { .flatten() } +/// Whether a stored entry key names an expired bucket at `block_time_ms`. +/// Keys without bucket-start semantics never expire. +fn entry_key_expired(transform: &TimeRangeTransform, entry_key: &[u8], block_time_ms: u64) -> bool { + entry_key_bucket_start(entry_key) + .is_some_and(|start| transform.bucket_expired(start, block_time_ms)) +} + /// Filter a derived time-range entry-key set down to the keys whose /// bucket has not expired at `block_time_ms`. Keys without bucket-start /// semantics (the null entry, raw keys) always pass; everything passes @@ -69,27 +95,50 @@ pub(crate) fn live_time_range_entry_keys( entry_keys: Vec>, block_time_ms: u64, ) -> Vec> { - let Some(horizon) = transform.expiry_horizon_ms(block_time_ms) else { - return entry_keys; - }; entry_keys .into_iter() - .filter(|key| entry_key_bucket_start(key).is_none_or(|start| start >= horizon)) + .filter(|key| !entry_key_expired(transform, key, block_time_ms)) .collect() } +/// [`live_time_range_entry_keys`] for the insert walker's key infos. +/// Size-only keys (the estimation dry run) always pass. +pub(crate) fn live_time_range_index_keys<'a>( + transform: &TimeRangeTransform, + index_keys: Vec>, + block_time_ms: u64, +) -> Vec> { + index_keys + .into_iter() + .filter(|key| match key { + DriveKeyInfo::Key(key) => !entry_key_expired(transform, key, block_time_ms), + DriveKeyInfo::KeyRef(key) => !entry_key_expired(transform, key, block_time_ms), + DriveKeyInfo::KeySize(_) => true, + }) + .collect() +} + +/// What a removal walker finds for a time-range entry key. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum TimeRangeEntryState { + /// A live (or non-bucket) key: remove as usual, no read performed. + Live, + /// The bucket is expired but still on disk — the window between + /// expiry and its lazy drop, where the entry must still be removed, + /// at full-path granularity (drainage may have taken its deeper + /// trees already). + ExpiredStanding, + /// The bucket is gone, and the entry with it: nothing to remove. + ExpiredGone, +} + impl Drive { - /// Whether a removal walker should process the time-range entry at - /// `entry_key` under the grid level at `level_path`. - /// - /// `true` for every live (or non-bucket) key with no read performed; - /// for an expired key, `true` exactly when the bucket value tree still - /// exists — the window between expiry and its lazy drop, where the - /// document's entries are still on disk and must still be removed. + /// Classifies the time-range entry at `entry_key` under the grid level + /// at `level_path` for a removal walker — see [`TimeRangeEntryState`]. /// Callers in estimation mode must not call this (state reads have no /// place in a dry run); they process every key, which keeps the dry /// run an upper bound. - pub(crate) fn time_range_entry_is_removable( + pub(crate) fn time_range_entry_state( &self, transform: &TimeRangeTransform, entry_key: &[u8], @@ -97,36 +146,33 @@ impl Drive { level_path: &[Vec], transaction: TransactionArg, platform_version: &PlatformVersion, - ) -> Result { + ) -> Result { + if !entry_key_expired(transform, entry_key, block_time_ms) { + return Ok(TimeRangeEntryState::Live); + } // Unbilled bookkeeping read — see the module's Billing section. let mut scratch_operations: Vec = vec![]; - let Some(horizon) = transform.expiry_horizon_ms(block_time_ms) else { - return Ok(true); - }; - let Some(start) = entry_key_bucket_start(entry_key) else { - return Ok(true); - }; - if start >= horizon { - return Ok(true); - } - let path_refs: Vec<&[u8]> = level_path - .iter() - .map(|segment| segment.as_slice()) - .collect(); - self.grove_has_raw( - SubtreePath::from(path_refs.as_slice()), + let standing = self.grove_has_raw( + SubtreePath::from(level_path), entry_key, DirectQueryType::StatefulDirectQuery, transaction, &mut scratch_operations, &platform_version.drive, - ) + )?; + Ok(if standing { + TimeRangeEntryState::ExpiredStanding + } else { + TimeRangeEntryState::ExpiredGone + }) } - /// Whether every segment of `path_segments` beyond the first - /// `known_prefix_len` (a prefix known to exist — the contract's - /// document-type path) resolves, walked one `has_raw` at a time so a + /// Whether every segment of `path_segments` from index + /// `known_prefix_len` on resolves, walked one `has_raw` at a time so a /// missing intermediate subtree answers `false` instead of erroring. + /// The prefix below `known_prefix_len` is known to exist — the callers + /// pass the index just past the bucket key, whose existence + /// [`Self::time_range_entry_state`] already established. /// /// The removal walkers use this at full-path granularity for entries /// in expired-but-standing buckets: TTL drainage removes whole `[0]` @@ -143,12 +189,8 @@ impl Drive { let mut scratch_operations: Vec = vec![]; let mut depth = known_prefix_len; while depth < path_segments.len() { - let parent_refs: Vec<&[u8]> = path_segments[..depth] - .iter() - .map(|segment| segment.as_slice()) - .collect(); if !self.grove_has_raw( - SubtreePath::from(parent_refs.as_slice()), + SubtreePath::from(&path_segments[..depth]), path_segments[depth].as_slice(), DirectQueryType::StatefulDirectQuery, transaction, @@ -162,7 +204,7 @@ impl Drive { Ok(true) } - /// Drain expired buckets from the grid level at `level_path` — the + /// Drain expired buckets from the grid level a request names — the /// lazy, budgeted cleanup every write into a TTL'd index continues. /// /// The drop primitive is grovedb's flat-subtree drop (grovedb#848 / @@ -185,14 +227,22 @@ impl Drive { /// per-axis secondary prefixes when it was an indexed primary; /// - the emptied bucket itself → flat drop. /// + /// The flat drops are batched: they are collected while the bucket is + /// walked and applied as ONE grovedb batch at the end, so a drain + /// costs a single root-hash propagation however many units it drops. + /// Only the indexed-tree deletes have no batched form; a node under a + /// ranked parent, and everything below it, is removed immediately + /// instead (the batch applies afterwards, so a deferred ancestor never + /// drops before an immediately removed descendant is gone). + /// /// Every step is a deterministic function of consensus state and block - /// time, and every step is O(1) — the *number* of steps is what scales - /// with user data (one per group, per level, per bucket), and that is - /// exactly what `max_operations` bounds per write. A bucket drains - /// across as many writes as it needs; between writes it stands - /// partially drained, which TTL semantics allow (entries live *at - /// most* `ttl`) and which the removal walkers handle at full-path - /// granularity. + /// time, and every step is O(1) in the subtree it drops — the *number* + /// of steps is what scales with user data (one per group, per level, + /// per bucket), and that is exactly what the request's + /// `max_operations` bounds per write and level. A bucket drains across + /// as many writes as it needs; between writes it stands partially + /// drained, which TTL semantics allow (entries live *at most* `ttl`) + /// and which the removal walkers handle at full-path granularity. /// /// The dropped paths embed their window start, so they are never /// re-created before their redo records drain (writes never target @@ -200,119 +250,96 @@ impl Drive { /// construction. The host completes reclamation by calling /// `GroveDb::flush_pending_prefix_drops` after committing the block's /// transaction (and once at startup). - /// One budgeted drainage pass for every TTL'd time-range level of a - /// document type. Levels are keyed by their grid-qualified storage key, - /// so indexes sharing a grid share one level and drain exactly once per - /// write — draining per *index* would multiply the per-write budget and, - /// worse, interleave direct grovedb drops with already-queued batch - /// mutations (a later index's drain can remove a path an earlier - /// index's pending operation targets). Callers must therefore run this - /// sweep BEFORE queuing any batch mutations, so every queued operation - /// describes post-drain state. Stateful only — never call from an - /// estimation dry run. - pub(crate) fn drain_expired_time_range_levels( - &self, - index_level: &IndexLevel, - contract_document_type_path: &[Vec], - block_time_ms: u64, - transaction: TransactionArg, - platform_version: &PlatformVersion, - ) -> Result<(), Error> { - let Some(max_operations) = platform_version - .system_limits - .max_time_range_ttl_drop_operations_per_write - else { - return Ok(()); - }; - for (name, sub_level) in index_level.sub_levels() { - if let Some(transform) = sub_level.time_range() { - if transform.ttl_seconds.is_some() { - let mut level_path = contract_document_type_path.to_vec(); - level_path.push(name.as_bytes().to_vec()); - self.drain_expired_time_range_buckets( - transform, - sub_level, - &level_path, - block_time_ms, - max_operations, - transaction, - platform_version, - )?; - } - } - } - Ok(()) - } - - #[allow(clippy::too_many_arguments)] pub(crate) fn drain_expired_time_range_buckets( &self, - transform: &TimeRangeTransform, - bucket_level: &IndexLevel, - level_path: &[Vec], - block_time_ms: u64, - max_operations: u16, + request: &TimeRangeTtlDrainRequest, transaction: TransactionArg, - platform_version: &PlatformVersion, + drive_version: &DriveVersion, ) -> Result<(), Error> { - let Some(horizon) = transform.expiry_horizon_ms(block_time_ms) else { + let TimeRangeTtlDrainRequest { + transform, + bucket_level, + level_path, + block_time_ms, + max_operations, + } = request; + let Some(horizon) = transform.expiry_horizon_ms(*block_time_ms) else { return Ok(()); }; // Unbilled system maintenance — see the module's Billing section. let drive_operations: &mut Vec = &mut vec![]; - let mut budget = max_operations; + let bucket_tree_type = + index_level_tree_types_with_continuation_demotion(bucket_level)?.value_tree_type; + let mut deferred_drops: Vec = vec![]; + let mut drained_buckets: Vec> = vec![]; + let mut budget = *max_operations; + + // Oldest expired bucket first. The null entry (empty key) sorts + // below every bucket start and is excluded — null entries are not + // windowed and live until their document goes. Only 8-byte keys + // carry bucket semantics. With today's grammar no other key can + // sort below the horizon anyway — the source is a required system + // timestamp, so the level holds 8-byte bucket starts plus at most + // the single null entry (empty key) — but the range start and the + // wider limit keep the finder live even if a future grammar admits + // raw (non-timestamp) keys: without them, low-sorting raw keys + // could fill every result slot and stall drainage forever. + let horizon_key = DocumentPropertyType::encode_date_timestamp(horizon); + let mut below_horizon = Query::new(); + below_horizon.insert_range(vec![0u8; 8]..horizon_key); + let path_query = PathQuery::new( + level_path.clone(), + SizedQuery::new( + below_horizon, + Some(max_operations.saturating_add(1).max(8)), + None, + ), + ); while budget > 0 { - // Oldest expired bucket first. The null entry (empty key) - // sorts below every bucket start and is excluded — null - // entries are not windowed and live until their document goes. - let horizon_key = DocumentPropertyType::encode_date_timestamp(horizon); - let mut below_horizon = Query::new(); - // Only 8-byte keys carry bucket semantics. With today's grammar - // no other key can sort below the horizon anyway — the source - // is a required system timestamp, so the level holds 8-byte - // bucket starts plus at most the single null entry (empty key) - // — but the range start and the wider limit keep the finder - // live even if a future grammar admits raw (non-timestamp) - // keys: without them, low-sorting raw keys could fill every - // result slot and stall drainage forever. - below_horizon.insert_range(vec![0u8; 8]..horizon_key); - let path_query = PathQuery::new( - level_path.to_vec(), - SizedQuery::new(below_horizon, Some(8), None), - ); let (results, _) = self.grove_get_raw_path_query( &path_query, transaction, QueryResultType::QueryKeyElementPairResultType, drive_operations, - &platform_version.drive, + drive_version, )?; - let Some(bucket_key) = results - .to_key_elements() - .into_iter() - .map(|(key, _)| key) - .find(|key| entry_key_bucket_start(key).is_some_and(|start| start < horizon)) - else { - return Ok(()); + // Deferred drops leave a fully drained bucket on disk until the + // batch applies, so skip the ones this drain already took. + let Some(bucket_key) = results.to_keys().into_iter().find(|key| { + entry_key_bucket_start(key).is_some_and(|start| start < horizon) + && !drained_buckets.contains(key) + }) else { + break; }; - let mut bucket_path = level_path.to_vec(); + let mut bucket_path = level_path.clone(); bucket_path.push(bucket_key.clone()); let fully_drained = self.drain_expired_node( &bucket_path, bucket_level, - level_path, - &bucket_key, + bucket_tree_type, // The grid level is never an indexed primary — ranking the // bucketed level is rejected at contract validation. TreeType::NormalTree, + false, &mut budget, + &mut deferred_drops, transaction, drive_operations, - platform_version, + drive_version, )?; if !fully_drained { - return Ok(()); + break; } + drained_buckets.push(bucket_key); + } + if !deferred_drops.is_empty() { + self.apply_batch_grovedb_operations( + None, + transaction, + GroveDbOpBatch::from_operations(deferred_drops), + drive_operations, + drive_version, + )?; } Ok(()) } @@ -320,33 +347,44 @@ impl Drive { /// Drain one tree of an expired bucket, deepest-first, then drop the /// tree itself. Returns whether the tree was fully removed (`false` ⇒ /// the budget ran out mid-way; the next write resumes exactly here, - /// because every completed step is a real removal). + /// because every completed step is a real removal once the deferred + /// batch applies). /// /// `level` describes the merged contract-known structure below this /// tree (property-name children by level key); the value-tree children - /// under each property-name tree are user data, enumerated one at a - /// time. + /// under each property-name tree are user data, enumerated once per + /// visit up to the remaining budget. #[allow(clippy::too_many_arguments)] fn drain_expired_node( &self, node_path: &[Vec], level: &IndexLevel, - parent_path: &[Vec], - node_key: &[u8], + node_tree_type: TreeType, parent_tree_type: TreeType, + parent_removed_immediately: bool, budget: &mut u16, + deferred_drops: &mut Vec, transaction: TransactionArg, drive_operations: &mut Vec, - platform_version: &PlatformVersion, + drive_version: &DriveVersion, ) -> Result { - let drive_version = &platform_version.drive; + let Some((node_key, parent_path)) = node_path.split_last() else { + return Err(Error::Drive(DriveError::CorruptedCodeExecution( + "a drained time-range node always has a parent", + ))); + }; + // A node under an indexed-primary parent leaves through grovedb's + // dedicated indexed-tree delete, which has no batched form and must + // find the node already empty — so it, and everything below it, is + // removed immediately. Every other unit is a deferred flat drop. + let removed_immediately = + parent_removed_immediately || is_indexed_primary(parent_tree_type); + // 1) Contract-known property-name children. for (level_key, sub_level) in level.sub_levels() { let level_key_bytes = level_key.as_bytes(); - let node_path_refs: Vec<&[u8]> = - node_path.iter().map(|segment| segment.as_slice()).collect(); let pn_element = self.grove_get_raw_optional( - SubtreePath::from(node_path_refs.as_slice()), + SubtreePath::from(node_path), level_key_bytes, DirectQueryType::StatefulDirectQuery, transaction, @@ -356,84 +394,93 @@ impl Drive { if pn_element.is_none() { continue; } - let pn_tree_type = index_level_tree_types_with_continuation_demotion(sub_level)? - .property_name_tree_type; + let sub_tree_types = index_level_tree_types_with_continuation_demotion(sub_level)?; let mut pn_path = node_path.to_vec(); pn_path.push(level_key_bytes.to_vec()); - // 1a) User-data value-tree children, one at a time. - loop { - if *budget == 0 { - return Ok(false); - } - let mut all = Query::new(); - all.insert_all(); - let path_query = - PathQuery::new(pn_path.clone(), SizedQuery::new(all, Some(1), None)); - let (results, _) = self.grove_get_raw_path_query( - &path_query, - transaction, - QueryResultType::QueryKeyElementPairResultType, - drive_operations, - drive_version, - )?; - let Some((value_key, _)) = results.to_key_elements().into_iter().next() else { - break; - }; + // 1a) User-data value-tree children, enumerated once up to the + // remaining budget: each costs at least one operation, so a + // full page means the budget is spent before the page is, and + // the next drain re-enumerates. + if *budget == 0 { + return Ok(false); + } + let mut all = Query::new(); + all.insert_all(); + let path_query = + PathQuery::new(pn_path.clone(), SizedQuery::new(all, Some(*budget), None)); + let (results, _) = self.grove_get_raw_path_query( + &path_query, + transaction, + QueryResultType::QueryKeyElementPairResultType, + drive_operations, + drive_version, + )?; + let value_keys = results.to_keys(); + let page_full = value_keys.len() == usize::from(*budget); + for value_key in value_keys { let mut value_path = pn_path.clone(); - value_path.push(value_key.clone()); + value_path.push(value_key); if !self.drain_expired_node( &value_path, sub_level, - &pn_path, - &value_key, - pn_tree_type, + sub_tree_types.value_tree_type, + sub_tree_types.property_name_tree_type, + removed_immediately, budget, + deferred_drops, transaction, drive_operations, - platform_version, + drive_version, )? { return Ok(false); } } + if page_full { + return Ok(false); + } // 1b) The drained property-name tree — flat drop, which also // dooms its per-axis secondary prefixes when it was indexed. if *budget == 0 { return Ok(false); } - self.grove_drop_flat_subtree( + self.drop_flat_unit( node_path, level_key_bytes, + sub_tree_types.property_name_tree_type, + removed_immediately, + deferred_drops, transaction, drive_operations, - platform_version, + drive_version, )?; *budget -= 1; } - // 2) The terminal `[0]` reference tree, when this level hosts one - // (non-unique / indexOnly layouts; the unique layout stores the - // reference AT key `[0]` as a bare element, which the flat drop of - // this node covers). - let node_path_refs: Vec<&[u8]> = - node_path.iter().map(|segment| segment.as_slice()).collect(); - let zero_element = self.grove_get_raw_optional( - SubtreePath::from(node_path_refs.as_slice()), - &[0], - DirectQueryType::StatefulDirectQuery, - transaction, - drive_operations, - drive_version, - )?; - if let Some(element) = zero_element { - if element.is_any_tree() { + // 2) The terminal `[0]` reference tree, when an index terminates at + // this level (non-unique / indexOnly layouts; the unique layout + // stores the reference AT key `[0]` as a bare element, which the + // flat drop of this node covers). + if let Some(index_type) = level.has_index_with_type() { + let zero_element = self.grove_get_raw_optional( + SubtreePath::from(node_path), + &[0], + DirectQueryType::StatefulDirectQuery, + transaction, + drive_operations, + drive_version, + )?; + if zero_element.is_some_and(|element| element.is_any_tree()) { if *budget == 0 { return Ok(false); } - self.grove_drop_flat_subtree( + self.drop_flat_unit( node_path, &[0], + terminal_member_tree_type(index_type), + removed_immediately, + deferred_drops, transaction, drive_operations, - platform_version, + drive_version, )?; *budget -= 1; } @@ -447,18 +494,14 @@ impl Drive { if *budget == 0 { return Ok(false); } - let parent_path_refs: Vec<&[u8]> = parent_path - .iter() - .map(|segment| segment.as_slice()) - .collect(); match parent_tree_type { TreeType::ProvableCountIndexedTree => { push_drive_operation_result( self.grove.delete_from_count_indexed_tree( - SubtreePath::from(parent_path_refs.as_slice()), + SubtreePath::from(parent_path), node_key, transaction, - &platform_version.drive.grove_version, + &drive_version.grove_version, ), drive_operations, )?; @@ -466,10 +509,10 @@ impl Drive { TreeType::ProvableSumIndexedTree => { push_drive_operation_result( self.grove.delete_from_provable_sum_indexed_tree( - SubtreePath::from(parent_path_refs.as_slice()), + SubtreePath::from(parent_path), node_key, transaction, - &platform_version.drive.grove_version, + &drive_version.grove_version, ), drive_operations, )?; @@ -478,21 +521,24 @@ impl Drive { push_drive_operation_result( self.grove .delete_from_provable_count_provable_sum_indexed_tree( - SubtreePath::from(parent_path_refs.as_slice()), + SubtreePath::from(parent_path), node_key, transaction, - &platform_version.drive.grove_version, + &drive_version.grove_version, ), drive_operations, )?; } _ => { - self.grove_drop_flat_subtree( + self.drop_flat_unit( parent_path, node_key, + node_tree_type, + removed_immediately, + deferred_drops, transaction, drive_operations, - platform_version, + drive_version, )?; } } @@ -500,6 +546,39 @@ impl Drive { Ok(true) } + /// One flat-drop unit of a drain: executed right away when the unit + /// sits under an immediately removed ancestor, deferred into the + /// drain's single batch otherwise. + #[allow(clippy::too_many_arguments)] + fn drop_flat_unit( + &self, + path: &[Vec], + key: &[u8], + tree_type: TreeType, + immediately: bool, + deferred_drops: &mut Vec, + transaction: TransactionArg, + drive_operations: &mut Vec, + drive_version: &DriveVersion, + ) -> Result<(), Error> { + if immediately { + return self.grove_drop_flat_subtree( + path, + key, + transaction, + drive_operations, + drive_version, + ); + } + deferred_drops.push(QualifiedGroveDbOp::delete_tree_op( + path.to_vec(), + key.to_vec(), + tree_type, + SubelementsDeletionBehavior::DropFlat, + )); + Ok(()) + } + /// Cost-pushing wrapper over [`GroveDb::drop_flat_subtree`] — the O(1) /// consensus detach of a flat subtree with staged prefix reclamation /// (grovedb#848). Version-gated inside grovedb itself: fail-closed @@ -511,15 +590,64 @@ impl Drive { key: &[u8], transaction: TransactionArg, drive_operations: &mut Vec, - platform_version: &PlatformVersion, + drive_version: &DriveVersion, ) -> Result<(), Error> { - let path_refs: Vec<&[u8]> = path.iter().map(|segment| segment.as_slice()).collect(); let cost_context = self.grove.drop_flat_subtree( - SubtreePath::from(path_refs.as_slice()), + SubtreePath::from(path), key, transaction, - &platform_version.drive.grove_version, + &drive_version.grove_version, ); push_drive_operation_result(cost_context, drive_operations) } } + +/// The three indexed-primary (ranked) tree types — the parents whose +/// children grovedb only removes through the dedicated indexed deletes. +fn is_indexed_primary(tree_type: TreeType) -> bool { + matches!( + tree_type, + TreeType::ProvableCountIndexedTree + | TreeType::ProvableSumIndexedTree + | TreeType::ProvableCountProvableSumIndexedTree + ) +} + +/// Queues one drainage request per TTL'd time-range level of a document +/// type — the sweep every write (insert, update or delete) into the type +/// performs. Levels are keyed by their grid-qualified storage key, so +/// indexes sharing a grid share one level and one request; requests for +/// the same level collapse again when +/// `apply_batch_low_level_drive_operations` runs them after the batch. +/// Stateful only — never call from an estimation dry run. +pub(crate) fn request_expired_time_range_drains( + index_level: &IndexLevel, + contract_document_type_path: &[Vec], + block_time_ms: u64, + platform_version: &PlatformVersion, + batch_operations: &mut Vec, +) { + let Some(max_operations) = platform_version + .system_limits + .max_time_range_ttl_drop_operations_per_write + else { + return; + }; + for (name, sub_level) in index_level.sub_levels() { + if let Some(transform) = sub_level.time_range() { + if transform.ttl_seconds.is_some() { + let mut level_path = contract_document_type_path.to_vec(); + level_path.push(name.as_bytes().to_vec()); + batch_operations.push(LowLevelDriveOperation::TimeRangeTtlDrain( + TimeRangeTtlDrainRequest { + transform: transform.clone(), + bucket_level: sub_level.clone(), + level_path, + block_time_ms, + max_operations, + }, + )); + } + } + } +} diff --git a/packages/rs-drive/src/drive/document/update/internal/update_document_for_contract_operations/v1/mod.rs b/packages/rs-drive/src/drive/document/update/internal/update_document_for_contract_operations/v1/mod.rs index 04a11442492..cfb51fd2056 100644 --- a/packages/rs-drive/src/drive/document/update/internal/update_document_for_contract_operations/v1/mod.rs +++ b/packages/rs-drive/src/drive/document/update/internal/update_document_for_contract_operations/v1/mod.rs @@ -2,7 +2,9 @@ use crate::drive::constants::CONTRACT_DOCUMENTS_PATH_HEIGHT; use crate::drive::document::index_level_tree_types::{ index_level_tree_types_with_continuation_demotion, IndexLevelTreeTypes, }; -use crate::drive::document::time_range_ttl::{entry_key_bucket_start, live_time_range_entry_keys}; +use crate::drive::document::time_range_ttl::{ + live_time_range_entry_keys, request_expired_time_range_drains, TimeRangeEntryState, +}; use crate::drive::document::{ make_document_reference, make_document_reference_with_sum_item, read_document_sum_contribution, }; @@ -300,33 +302,32 @@ impl Drive { // beneath a `ProvableCount*` / `ProvableSum*` parent — // diverging from the insert path (consensus break). let index_structure = document_type.index_structure(); + // Operations under TTL'd (ephemeral) levels — one vec for the whole + // walk, so the emptiness climb of one index sees the removals of + // another index sharing its level; re-tagged ephemeral after the + // loop. + let mut ephemeral_batch_operations: Vec = vec![]; // TTL drainage rides every write into a TTL'd index — updates - // included, mirroring the v2 insert walker: a bounded number of - // deepest-first drop operations against the oldest expired bucket, - // resuming wherever the previous write's budget ran out. One sweep - // over the deduplicated levels, BEFORE the per-index loop queues - // any batch mutation: drainage applies directly to grovedb, so a - // per-index drain could both multiply the per-write budget (several - // indexes may share one grid level) and remove paths an earlier - // index's queued operations target. Running it first also keeps the - // loop coherent with the drained state: if the drain takes a bucket - // this document's old entries lived in, the old-entry removable - // checks skip it. This path is stateful-only (estimation redirected - // to the insert walker above), and drainage is unbilled — see the - // ttl module's Billing section. + // included. One request per deduplicated level (several indexes may + // share one grid level), run once the transition's batch has + // applied — see `apply_batch_low_level_drive_operations` — so the + // removals queued below always target trees that still stand. This + // path is stateful-only (estimation redirected to the insert walker + // above), and drainage is unbilled — see the ttl module's Billing + // section. { let base_path: Vec> = contract_document_type_path .iter() .map(|&segment| Vec::from(segment)) .collect(); - self.drain_expired_time_range_levels( + request_expired_time_range_drains( index_structure, &base_path, block_info.time_ms, - transaction, platform_version, - )?; + &mut batch_operations, + ); } // fourth we need to store a reference to the document for each index @@ -389,14 +390,30 @@ impl Drive { // from ancestor sum aggregates (the document body remains // queryable but SUM/AVG proofs would exclude it — a soundness // bug an attacker could trigger with any benign no-op update). + // + // TTL'd (ephemeral) levels ride their own op batch and carry no + // storage flags — same routing as the insert and delete + // walkers; see the ttl module's Billing section. The reference + // is built with the level's flags: an ephemeral reference must + // be flagless or its later removal turns sectioned (refundable). + let index_is_ephemeral = current_index_level + .time_range() + .is_some_and(|transform| transform.ttl_seconds.is_some()); + let index_storage_flags = if index_is_ephemeral { + None + } else { + storage_flags + }; let index_document_reference = if let Some(sum_property_name) = &index.summable { let sum_value = read_document_sum_contribution(document, sum_property_name)?; make_document_reference_with_sum_item( document, document_and_contract_info.document_type, sum_value, - storage_flags, + index_storage_flags, ) + } else if index_is_ephemeral { + make_document_reference(document, document_and_contract_info.document_type, None) } else { document_reference.clone() }; @@ -413,45 +430,12 @@ impl Drive { // transform is exactly the grid every index sharing this level // declared.) if let Some(transform) = current_index_level.time_range() { - // TTL'd (ephemeral) sub-levels ride their own op batch and carry - // no storage flags — same routing as the insert and delete - // walkers; see the ttl module's Billing section. - let index_is_ephemeral = transform.ttl_seconds.is_some(); - let mut ephemeral_local_operations: Vec = vec![]; let index_batch_operations: &mut Vec = if index_is_ephemeral { - &mut ephemeral_local_operations + &mut ephemeral_batch_operations } else { &mut batch_operations }; - let index_storage_flags = if index_is_ephemeral { - None - } else { - storage_flags - }; - // The prebuilt reference bakes the document's flags into the - // element; ephemeral references must be flagless or their - // later removal turns sectioned (refundable). - let index_document_reference = if index_is_ephemeral { - if let Some(sum_property_name) = &index.summable { - let sum_value = - read_document_sum_contribution(document, sum_property_name)?; - make_document_reference_with_sum_item( - document, - document_and_contract_info.document_type, - sum_value, - None, - ) - } else { - make_document_reference( - document, - document_and_contract_info.document_type, - None, - ) - } - } else { - index_document_reference - }; self.update_time_range_index_for_contract_operations_v1( index, transform, @@ -470,13 +454,6 @@ impl Drive { transaction, platform_version, )?; - if index_is_ephemeral { - batch_operations.extend( - ephemeral_local_operations - .into_iter() - .map(LowLevelDriveOperation::retag_ephemeral), - ); - } continue; } @@ -949,6 +926,10 @@ impl Drive { } } } + LowLevelDriveOperation::push_retagged_ephemeral( + &mut batch_operations, + ephemeral_batch_operations, + ); Ok(batch_operations) } @@ -1013,11 +994,11 @@ impl Drive { ) -> Result<(), Error> { let drive_version = &platform_version.drive; - // TTL drainage already ran: the caller sweeps every TTL'd level - // once (deduplicated) before the per-index loop, so the removable - // checks below see post-drain state and no direct drop can race a - // queued mutation. Do not drain here — several indexes may share - // this level. + // TTL drainage is requested by the caller once per deduplicated + // level, before this per-index loop, and runs after the batch + // applies — so the removable checks below see standing state and no + // drop can race a queued mutation. Do not request it here — several + // indexes may share this level. // New/old raw values for the bucketed source property → entry key // sets, mirroring the insert walker's fan-out (see the doc comment). @@ -1356,38 +1337,35 @@ impl Drive { // bucket behaves exactly as before. This path is stateful-only // (estimation redirects to the insert walker at the top of the // v1 update), so the existence reads are always legal here. - let expired_entry = entry_key_bucket_start(entry_key) - .zip(transform.expiry_horizon_ms(block_time_ms)) - .is_some_and(|(start, horizon)| start < horizon); - if expired_entry { - if !self.time_range_entry_is_removable( - transform, - entry_key, - block_time_ms, - base_index_path, - transaction, - platform_version, - )? { - continue; - } - let mut entry_path_segments: Vec> = base_index_path.to_vec(); - entry_path_segments.push(entry_key.clone()); - for segment in &old_suffix { - entry_path_segments.push(segment.clone()); - } - if !old_terminator_is_unique { - entry_path_segments.push(vec![0]); - } - // `base_index_path` — the document-type path plus the - // grid-qualified level key — exists for every registered - // contract, so the walk starts below it. - if !self.expired_entry_path_exists( - &entry_path_segments, - base_index_path.len(), - transaction, - platform_version, - )? { - continue; + match self.time_range_entry_state( + transform, + entry_key, + block_time_ms, + base_index_path, + transaction, + platform_version, + )? { + TimeRangeEntryState::Live => {} + TimeRangeEntryState::ExpiredGone => continue, + TimeRangeEntryState::ExpiredStanding => { + let mut entry_path_segments: Vec> = base_index_path.to_vec(); + entry_path_segments.push(entry_key.clone()); + for segment in &old_suffix { + entry_path_segments.push(segment.clone()); + } + if !old_terminator_is_unique { + entry_path_segments.push(vec![0]); + } + // The bucket itself was just found standing, so the + // walk starts below it. + if !self.expired_entry_path_exists( + &entry_path_segments, + base_index_path.len() + 1, + transaction, + platform_version, + )? { + continue; + } } } let mut key_info_path: Vec = base_index_path diff --git a/packages/rs-drive/src/fees/op.rs b/packages/rs-drive/src/fees/op.rs index 8c0c2a72a74..5524470aa2f 100644 --- a/packages/rs-drive/src/fees/op.rs +++ b/packages/rs-drive/src/fees/op.rs @@ -5,6 +5,7 @@ use grovedb_costs::storage_cost::removal::StorageRemovedBytes::{ }; use std::collections::BTreeMap; +use dpp::data_contract::document_type::{IndexLevel, TimeRangeTransform}; use enum_map::Enum; use grovedb::batch::key_info::KeyInfo; use grovedb::batch::KeyInfoPath; @@ -20,7 +21,7 @@ use crate::error::Error; use crate::fees::get_overflow_error; use crate::fees::op::LowLevelDriveOperation::{ CalculatedCostOperation, CalculatedEphemeralCostOperation, EphemeralGroveOperation, - FunctionOperation, GroveOperation, PreCalculatedFeeResult, + FunctionOperation, GroveOperation, PreCalculatedFeeResult, TimeRangeTtlDrain, }; use crate::util::batch::grovedb_op_batch::GroveDbOpBatchV0Methods; use crate::util::storage_flags::StorageFlags; @@ -197,6 +198,32 @@ impl FunctionOp { } } +/// A deferred TTL drainage request for one time-range index level. +/// +/// The document walkers emit one per write into a TTL'd `timeRange` +/// level instead of draining inline: drainage performs real grovedb +/// removals, and running it while a transition's other operations were +/// still queued let it take subtrees those pending operations targeted. +/// `apply_batch_low_level_drive_operations` runs the requests once the +/// batch is applied, one drain per level (requests for the same level +/// collapse), so the per-write budget holds per level however many +/// indexes share it. +#[derive(Debug, Clone, PartialEq)] +pub struct TimeRangeTtlDrainRequest { + /// The level's transform (carries the ttl). + pub transform: TimeRangeTransform, + /// The merged contract structure below the level's buckets. + pub bucket_level: IndexLevel, + /// The level path: the document-type path plus the grid-qualified key. + pub level_path: Vec>, + /// Block time the expiry horizon derives from. + pub block_time_ms: u64, + /// The drop budget (`SystemLimits::max_time_range_ttl_drop_operations_per_write`). + pub max_operations: u16, +} + +impl Eq for TimeRangeTtlDrainRequest {} + /// Drive operation // GroveOperation dominates every op vec on the write path; boxing it would // trade one inline copy for a per-op heap allocation in consensus-critical @@ -226,6 +253,10 @@ pub enum LowLevelDriveOperation { CalculatedEphemeralCostOperation(OperationCost), /// Pre Calculated Fee Result PreCalculatedFeeResult(FeeResult), + /// A deferred TTL drainage request, executed by + /// `apply_batch_low_level_drive_operations` after the batch is applied + /// — see [`TimeRangeTtlDrainRequest`]. Never reaches fee consumption. + TimeRangeTtlDrain(TimeRangeTtlDrainRequest), } /// Shared rejection message for the three `Element` wrappers @@ -382,42 +413,62 @@ impl LowLevelDriveOperation { FunctionOperation(_) => Err(Error::Drive(DriveError::CorruptedCodeExecution( "function operations should not be requested by operation costs", ))), + TimeRangeTtlDrain(_) => Err(Error::Drive(DriveError::CorruptedCodeExecution( + "time-range ttl drain requests must be executed by \ + apply_batch_low_level_drive_operations, not transformed to costs", + ))), } } - /// Filters the groveDB ops from a list of operations and puts them in a `GroveDbOpBatch`. + /// Sums the calculated costs of a list of operations, both pricing + /// classes included. pub fn combine_cost_operations(operations: &[LowLevelDriveOperation]) -> OperationCost { let mut cost = OperationCost::default(); operations.iter().for_each(|op| { - if let CalculatedCostOperation(operation_cost) = op { + if let CalculatedCostOperation(operation_cost) + | CalculatedEphemeralCostOperation(operation_cost) = op + { cost += operation_cost.clone() } }); cost } + /// The grove operation this op carries, whichever pricing class it is + /// tagged with. + pub fn grove_op_ref(&self) -> Option<&QualifiedGroveDbOp> { + match self { + GroveOperation(grovedb_op) | EphemeralGroveOperation(grovedb_op) => Some(grovedb_op), + _ => None, + } + } + /// Filters the groveDB ops from a list of operations and puts them in a `GroveDbOpBatch`. + /// + /// Ephemeral (TTL'd-subtree) ops are included so no write is ever + /// dropped; only [`Self::grovedb_operations_batch_consume_split_ephemeral`] + /// keeps the two pricing classes apart. pub fn grovedb_operations_batch( insert_operations: &[LowLevelDriveOperation], ) -> GroveDbOpBatch { let operations = insert_operations .iter() - .filter_map(|op| match op { - GroveOperation(grovedb_op) => Some(grovedb_op.clone()), - _ => None, - }) + .filter_map(|op| op.grove_op_ref().cloned()) .collect(); GroveDbOpBatch::from_operations(operations) } /// Filters the groveDB ops from a list of operations and puts them in a `GroveDbOpBatch`. + /// Ephemeral ops are included — see [`Self::grovedb_operations_batch`]. pub fn grovedb_operations_batch_consume( insert_operations: Vec, ) -> GroveDbOpBatch { let operations = insert_operations .into_iter() .filter_map(|op| match op { - GroveOperation(grovedb_op) => Some(grovedb_op), + GroveOperation(grovedb_op) | EphemeralGroveOperation(grovedb_op) => { + Some(grovedb_op) + } _ => None, }) .collect(); @@ -425,15 +476,13 @@ impl LowLevelDriveOperation { } /// Filters the groveDB ops from a list of operations and puts them in a `GroveDbOpBatch`. + /// Ephemeral ops are folded into the batch — see [`Self::grovedb_operations_batch`]. pub fn grovedb_operations_batch_consume_with_leftovers( insert_operations: Vec, ) -> (GroveDbOpBatch, Vec) { - let (batch, ephemeral_batch, other_operations) = + let (mut batch, ephemeral_batch, other_operations) = Self::grovedb_operations_batch_consume_split_ephemeral(insert_operations); - debug_assert!( - ephemeral_batch.is_empty(), - "ephemeral grove operations must go through the ephemeral-aware apply" - ); + batch.operations.extend(ephemeral_batch.operations); (batch, other_operations) } @@ -474,14 +523,40 @@ impl LowLevelDriveOperation { } } + /// Appends `local` to `batch_operations`, re-tagged ephemeral. + pub fn push_retagged_ephemeral(batch_operations: &mut Vec, local: Vec) { + batch_operations.extend(local.into_iter().map(Self::retag_ephemeral)); + } + + /// Runs `f` against the vec an index sub-level's operations belong in: + /// `batch_operations` itself for a standing level, a local vec that is + /// re-tagged ephemeral on the way out for a TTL'd one. The single + /// spelling of the walkers' ephemeral routing. + pub fn with_ephemeral_routing( + batch_operations: &mut Vec, + is_ephemeral: bool, + f: impl FnOnce(&mut Vec) -> Result, + ) -> Result { + if !is_ephemeral { + return f(batch_operations); + } + let mut local = vec![]; + let result = f(&mut local)?; + Self::push_retagged_ephemeral(batch_operations, local); + Ok(result) + } + /// Filters the groveDB ops from a list of operations and collects them in a `Vec`. + /// Ephemeral ops are included — see [`Self::grovedb_operations_batch`]. pub fn grovedb_operations_consume( insert_operations: Vec, ) -> Vec { insert_operations .into_iter() .filter_map(|op| match op { - GroveOperation(grovedb_op) => Some(grovedb_op), + GroveOperation(grovedb_op) | EphemeralGroveOperation(grovedb_op) => { + Some(grovedb_op) + } _ => None, }) .collect() diff --git a/packages/rs-drive/src/util/grove_operations/batch_insert_empty_tree_if_not_exists/v0/mod.rs b/packages/rs-drive/src/util/grove_operations/batch_insert_empty_tree_if_not_exists/v0/mod.rs index 4e425c7296c..5bfc3e4bae4 100644 --- a/packages/rs-drive/src/util/grove_operations/batch_insert_empty_tree_if_not_exists/v0/mod.rs +++ b/packages/rs-drive/src/util/grove_operations/batch_insert_empty_tree_if_not_exists/v0/mod.rs @@ -2,7 +2,6 @@ use super::EmptyTreeInsertMode; use crate::drive::Drive; use crate::error::drive::DriveError; use crate::error::Error; -use crate::fees::op::LowLevelDriveOperation::GroveOperation; use crate::fees::op::{LowLevelDriveOperation, LowLevelDriveOperationTreeTypeConverter}; use crate::util::grove_operations::BatchInsertTreeApplyType; use crate::util::object_size_info::PathKeyInfo; @@ -108,10 +107,11 @@ impl Drive { // if it already exists then just ignore things // if we had a delete then we need to remove the delete let previous_drive_operation = &existing_operations[i]; - if previous_drive_operation == &drive_operation { + if previous_drive_operation.grove_op_ref() == drive_operation.grove_op_ref() + { found = true; break; - } else if let GroveOperation(grove_op) = previous_drive_operation { + } else if let Some(grove_op) = previous_drive_operation.grove_op_ref() { if grove_op.key == Some(KeyInfo::KnownKey(key.to_vec())) && grove_op.path == path && matches!(grove_op.op, GroveOp::DeleteTree(_, _)) @@ -171,10 +171,11 @@ impl Drive { // if it already exists then just ignore things // if we had a delete then we need to remove the delete let previous_drive_operation = &existing_operations[i]; - if previous_drive_operation == &drive_operation { + if previous_drive_operation.grove_op_ref() == drive_operation.grove_op_ref() + { found = true; break; - } else if let GroveOperation(grove_op) = previous_drive_operation { + } else if let Some(grove_op) = previous_drive_operation.grove_op_ref() { if grove_op.key == Some(KeyInfo::KnownKey(key.to_vec())) && grove_op.path == path && matches!(grove_op.op, GroveOp::DeleteTree(_, _)) @@ -232,10 +233,11 @@ impl Drive { // if it already exists then just ignore things // if we had a delete then we need to remove the delete let previous_drive_operation = &existing_operations[i]; - if previous_drive_operation == &drive_operation { + if previous_drive_operation.grove_op_ref() == drive_operation.grove_op_ref() + { found = true; break; - } else if let GroveOperation(grove_op) = previous_drive_operation { + } else if let Some(grove_op) = previous_drive_operation.grove_op_ref() { if grove_op.key == Some(KeyInfo::KnownKey(key.to_vec())) && grove_op.path == path && matches!(grove_op.op, GroveOp::DeleteTree(_, _)) @@ -293,10 +295,11 @@ impl Drive { // if it already exists then just ignore things // if we had a delete then we need to remove the delete let previous_drive_operation = &existing_operations[i]; - if previous_drive_operation == &drive_operation { + if previous_drive_operation.grove_op_ref() == drive_operation.grove_op_ref() + { found = true; break; - } else if let GroveOperation(grove_op) = previous_drive_operation { + } else if let Some(grove_op) = previous_drive_operation.grove_op_ref() { if grove_op.key == Some(KeyInfo::KnownKey(key.to_vec())) && grove_op.path == path && matches!(grove_op.op, GroveOp::DeleteTree(_, _)) diff --git a/packages/rs-drive/src/util/operations/apply_batch_low_level_drive_operations/v0/mod.rs b/packages/rs-drive/src/util/operations/apply_batch_low_level_drive_operations/v0/mod.rs index da56d5bc958..d750bbf1ccf 100644 --- a/packages/rs-drive/src/util/operations/apply_batch_low_level_drive_operations/v0/mod.rs +++ b/packages/rs-drive/src/util/operations/apply_batch_low_level_drive_operations/v0/mod.rs @@ -20,7 +20,7 @@ impl Drive { drive_operations: &mut Vec, drive_version: &DriveVersion, ) -> Result<(), Error> { - let (grove_db_operations, ephemeral_grove_db_operations, mut other_operations) = + let (grove_db_operations, ephemeral_grove_db_operations, other_operations) = LowLevelDriveOperation::grovedb_operations_batch_consume_split_ephemeral( batch_operations, ); @@ -60,6 +60,25 @@ impl Drive { .map(LowLevelDriveOperation::retag_ephemeral), ); } + // Deferred TTL drainage runs only now, with every operation of the + // transition applied: draining earlier could remove subtrees that + // queued removals still targeted. One drain per level — requests + // from indexes sharing a level collapse, so the per-write budget is + // spent once per level. Unbilled, like the drain itself. + let (drain_requests, mut other_operations): (Vec<_>, Vec<_>) = other_operations + .into_iter() + .partition(|op| matches!(op, LowLevelDriveOperation::TimeRangeTtlDrain(_))); + let mut drained_levels: Vec>> = vec![]; + for request in drain_requests { + let LowLevelDriveOperation::TimeRangeTtlDrain(request) = request else { + continue; + }; + if drained_levels.contains(&request.level_path) { + continue; + } + self.drain_expired_time_range_buckets(&request, transaction, drive_version)?; + drained_levels.push(request.level_path); + } drive_operations.append(&mut other_operations); Ok(()) } diff --git a/packages/rs-platform-version/src/version/fee/mod.rs b/packages/rs-platform-version/src/version/fee/mod.rs index c9d57857ba4..23b8be0bf25 100644 --- a/packages/rs-platform-version/src/version/fee/mod.rs +++ b/packages/rs-platform-version/src/version/fee/mod.rs @@ -11,7 +11,7 @@ use crate::version::fee::signature::FeeSignatureVersion; use crate::version::fee::state_transition_min_fees::{ StateTransitionMinFees, StateTransitionMinFeesBeforeProtocolVersion11, }; -use crate::version::fee::storage::FeeStorageVersion; +use crate::version::fee::storage::{FeeStorageVersion, FeeStorageVersionBeforeVersion14}; use crate::version::fee::v1::FEE_VERSION1; use crate::version::fee::vote_resolution_fund_fees::VoteResolutionFundFees; use bincode::{Decode, Encode}; @@ -90,7 +90,7 @@ impl FeeVersion { pub struct FeeVersionFieldsBeforeVersion4 { // Permille means devise by 1000 pub uses_version_fee_multiplier_permille: Option, - pub storage: FeeStorageVersion, + pub storage: FeeStorageVersionBeforeVersion14, pub signature: FeeSignatureVersion, pub hashing: FeeHashingVersionBeforeVersion11, pub processing: FeeProcessingVersionFieldsBeforeVersion1Point4, @@ -104,7 +104,7 @@ impl From for FeeVersion { FeeVersion { fee_version_number: 1, uses_version_fee_multiplier_permille: value.uses_version_fee_multiplier_permille, - storage: value.storage, + storage: FeeStorageVersion::from(value.storage), signature: value.signature, hashing: FEE_HASHING_VERSION1, processing: FeeProcessingVersion::from(value.processing), diff --git a/packages/rs-platform-version/src/version/fee/storage/mod.rs b/packages/rs-platform-version/src/version/fee/storage/mod.rs index 3846b9aed1b..e6f439626d3 100644 --- a/packages/rs-platform-version/src/version/fee/storage/mod.rs +++ b/packages/rs-platform-version/src/version/fee/storage/mod.rs @@ -20,9 +20,35 @@ pub struct FeeStorageVersion { pub ttl_ephemeral_disk_usage_credit_per_byte: u64, } +/// Frozen pre-protocol-version-14 layout of `FeeStorageVersion`, kept only so +/// that `FeeVersionFieldsBeforeVersion4` (persisted inside platform state +/// before 1.4) keeps decoding after new fields are added to the live struct. +#[derive(Clone, Debug, Encode, Decode, Default, PartialEq, Eq)] +pub struct FeeStorageVersionBeforeVersion14 { + pub storage_disk_usage_credit_per_byte: u64, + pub storage_processing_credit_per_byte: u64, + pub storage_load_credit_per_byte: u64, + pub non_storage_load_credit_per_byte: u64, + pub storage_seek_cost: u64, +} + +impl From for FeeStorageVersion { + fn from(value: FeeStorageVersionBeforeVersion14) -> Self { + FeeStorageVersion { + storage_disk_usage_credit_per_byte: value.storage_disk_usage_credit_per_byte, + storage_processing_credit_per_byte: value.storage_processing_credit_per_byte, + storage_load_credit_per_byte: value.storage_load_credit_per_byte, + non_storage_load_credit_per_byte: value.non_storage_load_credit_per_byte, + storage_seek_cost: value.storage_seek_cost, + // Unreachable before protocol version 14: the `ttl` grammar did not exist. + ttl_ephemeral_disk_usage_credit_per_byte: 0, + } + } +} + #[cfg(test)] mod tests { - use super::FeeStorageVersion; + use super::{FeeStorageVersion, FeeStorageVersionBeforeVersion14}; #[test] // If this test failed, then a new field was added in FeeProcessingVersion. And the corresponding eq needs to be updated as well @@ -48,4 +74,29 @@ mod tests { // This assertion will check if all fields are considered in the equality comparison assert_eq!(version1, version2, "FeeStorageVersion equality test failed. If a field was added or removed, update the Eq implementation."); } + + #[test] + // Guards the wire layout FeeVersionFieldsBeforeVersion4 persists: five u64s, nothing more. + fn test_fee_storage_version_before_version_14_decodes_pre_existing_layout() { + let frozen = FeeStorageVersionBeforeVersion14 { + storage_disk_usage_credit_per_byte: 1, + storage_processing_credit_per_byte: 2, + storage_load_credit_per_byte: 3, + non_storage_load_credit_per_byte: 4, + storage_seek_cost: 5, + }; + + let bytes = bincode::encode_to_vec(&frozen, bincode::config::standard()) + .expect("expected to encode frozen storage fee version"); + assert_eq!(bytes, vec![1, 2, 3, 4, 5]); + + let (decoded, _): (FeeStorageVersionBeforeVersion14, _) = + bincode::decode_from_slice(&bytes, bincode::config::standard()) + .expect("expected to decode frozen storage fee version"); + assert_eq!(decoded, frozen); + + let live = FeeStorageVersion::from(decoded); + assert_eq!(live.storage_seek_cost, 5); + assert_eq!(live.ttl_ephemeral_disk_usage_credit_per_byte, 0); + } } diff --git a/packages/rs-platform-version/src/version/system_limits/mod.rs b/packages/rs-platform-version/src/version/system_limits/mod.rs index f73ebfd18d1..9cc96691055 100644 --- a/packages/rs-platform-version/src/version/system_limits/mod.rs +++ b/packages/rs-platform-version/src/version/system_limits/mod.rs @@ -113,7 +113,8 @@ pub struct SystemLimits { /// `ttl` key (nothing to bound: the key does not parse there). pub max_time_range_ttl_seconds: Option, /// Maximum number of O(1) drop operations one write into a TTL'd - /// `timeRange` index may spend draining expired buckets. + /// `timeRange` index may spend draining expired buckets — per index + /// level: indexes sharing a level share one drain per transition. /// /// A bucket drains deepest-first through flat-subtree drops (one per /// `[0]` reference tree, per emptied value tree, per property-name diff --git a/packages/rs-platform-version/src/version/system_limits/v5.rs b/packages/rs-platform-version/src/version/system_limits/v5.rs index 375b12ab67b..6ff423cb015 100644 --- a/packages/rs-platform-version/src/version/system_limits/v5.rs +++ b/packages/rs-platform-version/src/version/system_limits/v5.rs @@ -12,9 +12,10 @@ use crate::version::system_limits::SystemLimits; /// honest price for transitional storage while the lifetime it covers is /// bounded. See `book/src/drive/time-range-ttl.md`. /// * `max_time_range_ttl_drop_operations_per_write` is set to 8: each -/// write into a TTL'd index spends at most this many O(1) flat-drop -/// operations draining expired buckets, deepest-first, resuming across -/// writes. +/// write into a TTL'd index level spends at most this many O(1) +/// flat-drop operations draining its expired buckets, deepest-first, +/// resuming across writes (once per level and transition, however many +/// indexes share the level). /// /// The changes carried over from the folded-in V4: ///