Repository navigation
feat(router): approx indexer support per-insert pruning TTLs #12793
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -116,26 +116,26 @@ impl<K: Clone + Hash + Eq + Ord> PruneManager<K> { | |
| /// Inserts a new timer or updates an existing one for the given key. | ||
| /// | ||
| /// # Arguments | ||
| /// * `keys` - The unique keys for the timers. | ||
| pub fn insert(&mut self, keys: Vec<K>) { | ||
| self.insert_at(keys, Instant::now(), None); | ||
| } | ||
|
|
||
| /// Inserts timers using a caller-provided timestamp. | ||
| pub fn insert_at(&mut self, keys: Vec<K>, now: Instant, ttl_override: Option<Duration>) { | ||
| let len = keys.len(); | ||
| let ttl = ttl_override.unwrap_or(self.ttl); | ||
| let expiry_time = if ttl.is_zero() { | ||
| now | ||
| } else { | ||
| self.bucket_expiry(now + ttl) | ||
| }; | ||
| let mut bucket_inserts: Option<Vec<K>> = None; | ||
|
|
||
| self.timers.reserve(len); | ||
| for key in keys { | ||
| match self.timers.entry(key.clone()) { | ||
| Entry::Occupied(entry) if *entry.get() >= expiry_time => { | ||
|
Contributor
There was a problem hiding this comment. AI review — human evaluation requested: Please re-evaluate the previously discussed stale-removal race against #12785's acceptance criterion. This max-expiry check only helps while the key remains in the timer map. Once |
||
| continue; | ||
| } | ||
| Entry::Occupied(mut entry) => { | ||
|
|
@@ -231,8 +231,13 @@ impl WorkerPruneState { | |
| } | ||
| } | ||
|
|
||
| fn insert_block_entries( | ||
| &mut self, | ||
| entries: Vec<BlockEntry>, | ||
| now: Instant, | ||
| ttl_override: Option<Duration>, | ||
| ) { | ||
| self.timers.insert_at(entries, now, ttl_override); | ||
| } | ||
|
|
||
| fn remove_block_entry(&mut self, entry: &BlockEntry) { | ||
|
|
@@ -318,7 +323,7 @@ impl WorkerPruneManager { | |
| } | ||
| } | ||
|
|
||
| pub fn insert_block_entries(&self, entries: Vec<BlockEntry>, ttl_override: Option<Duration>) { | ||
| if entries.is_empty() { | ||
| return; | ||
| } | ||
|
|
@@ -333,20 +338,25 @@ impl WorkerPruneManager { | |
| let mut should_bump_schedule = false; | ||
| for (worker, worker_entries) in by_worker { | ||
| should_bump_schedule |= | ||
| self.insert_worker_block_entries_at(worker, worker_entries, now, ttl_override); | ||
| } | ||
|
|
||
| if should_bump_schedule { | ||
| self.inner.bump_schedule(); | ||
| } | ||
| } | ||
|
|
||
| pub fn insert_worker_block_entries( | ||
| &self, | ||
| worker: WorkerWithDpRank, | ||
| entries: Vec<BlockEntry>, | ||
| ttl_override: Option<Duration>, | ||
| ) { | ||
| if entries.is_empty() { | ||
| return; | ||
| } | ||
|
|
||
| if self.insert_worker_block_entries_at(worker, entries, Instant::now(), ttl_override) { | ||
| self.inner.bump_schedule(); | ||
| } | ||
| } | ||
|
|
@@ -356,6 +366,7 @@ impl WorkerPruneManager { | |
| worker: WorkerWithDpRank, | ||
| entries: Vec<BlockEntry>, | ||
| now: Instant, | ||
| ttl_override: Option<Duration>, | ||
| ) -> bool { | ||
| let (old_next, new_next) = { | ||
| let state = | ||
|
|
@@ -364,28 +375,26 @@ impl WorkerPruneManager { | |
| }); | ||
| let mut state = state.lock().expect("worker prune state mutex poisoned"); | ||
| let old_next = state.peek_next_valid_expiry(); | ||
| state.insert_block_entries(entries, now, ttl_override); | ||
| let new_next = state.peek_next_valid_expiry(); | ||
| (old_next, new_next) | ||
| }; | ||
|
|
||
| self.update_worker_expiry(worker, old_next, new_next) | ||
| } | ||
|
|
||
| fn update_worker_expiry( | ||
| &self, | ||
| worker: WorkerWithDpRank, | ||
| old_next: Option<Instant>, | ||
| new_next: Option<Instant>, | ||
| ) -> bool { | ||
| match (old_next, new_next) { | ||
| (None, Some(next_expiry)) => { | ||
| self.inner.push_worker_expiry(worker, next_expiry); | ||
| true | ||
| } | ||
| (Some(old_next), Some(new_next)) if new_next < old_next => { | ||
| self.inner.push_worker_expiry(worker, new_next); | ||
|
Contributor
There was a problem hiding this comment. AI review — human evaluation requested: Please verify the interaction with concurrent heap rebuilds. A rebuild snapshots per-worker minima without serializing against this push, then replaces the heap wholesale. An insertion can publish a newly earlier minimum after its worker was snapshotted, only for the rebuild to overwrite and lose that entry; if another worker remains heap-top, stale-top validation does not rediscover the omitted worker until later. Heterogeneous TTLs make |
||
| true | ||
| } | ||
|
|
@@ -666,7 +675,7 @@ mod tests { | |
| let mut pm: PruneManager<u32> = PruneManager::new(prune_config); | ||
|
|
||
| let now = pm.bucket_origin + Duration::from_millis(1); | ||
| pm.insert_at(vec![1, 2, 3], now, None); | ||
| let expiry = *pm.get_expiry(&1).expect("expiry missing after insert"); | ||
| assert_eq!(pm.get_expiry(&2), Some(&expiry)); | ||
| assert_eq!(pm.get_expiry(&3), Some(&expiry)); | ||
|
|
@@ -686,8 +695,8 @@ mod tests { | |
| let mut pm: PruneManager<u32> = PruneManager::new(prune_config); | ||
|
|
||
| let now = pm.bucket_origin + Duration::from_millis(1); | ||
| pm.insert_at(vec![1], now, None); | ||
| pm.insert_at(vec![2], now + Duration::from_millis(20), None); | ||
|
|
||
| let expiry = *pm.get_expiry(&1).expect("expiry missing for key 1"); | ||
| assert_eq!(pm.get_expiry(&2), Some(&expiry)); | ||
|
|
@@ -706,10 +715,10 @@ mod tests { | |
| let mut pm: PruneManager<u32> = PruneManager::new(prune_config); | ||
|
|
||
| let now = pm.bucket_origin + Duration::from_millis(1); | ||
| pm.insert_at(vec![42], now, None); | ||
| let expiry = *pm.get_expiry(&42).expect("expiry missing for key 42"); | ||
|
|
||
| pm.insert_at(vec![42], now + Duration::from_millis(20), None); | ||
|
|
||
| assert_eq!(pm.get_expiry(&42), Some(&expiry)); | ||
| assert_eq!(pm.expirations.len(), 1); | ||
|
|
@@ -726,7 +735,7 @@ mod tests { | |
| let mut pm: PruneManager<u32> = PruneManager::new(prune_config); | ||
|
|
||
| let now = pm.bucket_origin + Duration::from_millis(1); | ||
| pm.insert_at(vec![7], now, None); | ||
|
|
||
| assert_eq!(pm.get_expiry(&7), Some(&now)); | ||
| assert_eq!(pm.pop_expired(now), vec![7]); | ||
|
|
@@ -741,12 +750,12 @@ mod tests { | |
| let mut pm: PruneManager<u32> = PruneManager::new(prune_config); | ||
|
|
||
| let now = pm.bucket_origin + Duration::from_millis(1); | ||
| pm.insert_at(vec![42], now, None); | ||
| let first_expiry = *pm | ||
| .get_expiry(&42) | ||
| .expect("expiry missing after first insert"); | ||
|
|
||
| pm.insert_at(vec![42], now + EXPIRY_BUCKET, None); | ||
| let second_expiry = *pm | ||
| .get_expiry(&42) | ||
| .expect("expiry missing after reinsertion"); | ||
|
|
@@ -760,6 +769,28 @@ mod tests { | |
| assert_eq!(expired_after, vec![42]); | ||
| } | ||
|
|
||
| #[test] | ||
| fn test_prune_manager_ttl_override_keeps_latest_expiry() { | ||
| let mut pm = PruneManager::new(PruneConfig { | ||
| ttl: Duration::from_secs(120), | ||
| }); | ||
| let now = pm.bucket_origin + Duration::from_millis(1); | ||
|
|
||
| pm.insert_at(vec![1], now, None); | ||
| pm.insert_at(vec![2], now, Some(Duration::from_secs(120))); | ||
| let default_expiry = *pm.get_expiry(&1).expect("default expiry missing"); | ||
| assert_eq!(pm.get_expiry(&2), Some(&default_expiry)); | ||
|
|
||
| pm.insert_at(vec![3], now, Some(Duration::from_secs(5))); | ||
| pm.insert_at(vec![2], now, Some(Duration::from_secs(300))); | ||
| let short_expiry = *pm.get_expiry(&3).expect("short expiry missing"); | ||
| let long_expiry = *pm.get_expiry(&2).expect("long expiry missing"); | ||
| pm.insert_at(vec![2], now, Some(Duration::from_secs(5))); | ||
|
|
||
| assert!(short_expiry < long_expiry); | ||
|
Contributor
There was a problem hiding this comment. AI review — human evaluation requested: This checks the leaf map's computed expiries, but it does not demonstrate that the background scheduler wakes for a newly earlier deadline. The interface-level test later calls |
||
| assert_eq!(pm.get_expiry(&2), Some(&long_expiry)); | ||
| } | ||
|
|
||
| #[test] | ||
| fn test_worker_expiry_heap_rebuilds_under_churn() { | ||
| let manager = WorkerPruneManager::new(PruneConfig { | ||
|
|
@@ -768,7 +799,7 @@ mod tests { | |
| let worker = WorkerWithDpRank::from_worker_id(7); | ||
|
|
||
| for idx in 0..=WORKER_EXPIRY_HEAP_REBUILD_THRESHOLD { | ||
| manager.insert_block_entries(vec![test_block(worker, 100 + idx as u64, idx)], None); | ||
| } | ||
|
|
||
| assert_eq!(manager.worker_expiry_heap_len(), 1); | ||
|
|
@@ -785,9 +816,9 @@ mod tests { | |
| let first = test_block(first_worker, 101, 0); | ||
| let second = test_block(second_worker, 202, 0); | ||
|
|
||
| manager.insert_block_entries(vec![first], None); | ||
| time::advance(Duration::from_secs(5)).await; | ||
| manager.insert_block_entries(vec![second], None); | ||
|
|
||
| time::advance(Duration::from_secs(5)).await; | ||
| assert_eq!(manager.drain_due_and_pending(Instant::now()), vec![first]); | ||
|
|
@@ -804,9 +835,9 @@ mod tests { | |
| let worker = WorkerWithDpRank::from_worker_id(7); | ||
| let block = test_block(worker, 707, 0); | ||
|
|
||
| manager.insert_block_entries(vec![block], None); | ||
| time::advance(Duration::from_secs(5)).await; | ||
| manager.insert_block_entries(vec![block], None); | ||
|
|
||
| time::advance(Duration::from_secs(5)).await; | ||
| assert!(manager.drain_due_and_pending(Instant::now()).is_empty()); | ||
|
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
AI review — human evaluation requested: This carries a relative TTL through the queue/RPC, but the pruning manager samples
Instant::now()only when it later processes the request. Local backpressure, a worker acknowledgement, or remote transport therefore does not consume the requested lifetime; a request with five seconds remaining after ten seconds in transit receives another five seconds. If the intended contract is event-age-bounded retention, consider computing a validated absolute internal deadline at admission and carrying that through private queues rather than restarting a relative duration at each boundary.