1#[cfg(test)]
2use std::cell::Cell;
3use std::{
4 collections::{BTreeMap, BTreeSet},
5 fmt,
6 sync::{Arc, Mutex},
7};
8
9use omena_reactive::{
10 ChangePolicyV0, ReactiveEngineV0, ReactiveGraphBuilderV0, ReactiveNodeIdV0, ReactiveStateV0,
11 ReactiveValueV0, StabilizeStatusV0,
12};
13
14const STABILIZATION_RECOMPUTE_LIMIT: usize = 64;
15const MODULE_INTERFACE_MEMO_ENTRY_LIMIT: usize = 2_048;
16const DELIVERY_EFFECT_CHANNEL: &str = "lspDiagnosticsDeliveryDecision";
17pub const REACTIVE_SHADOW_ENV: &str = "OMENA_LSP_REACTIVE_SHADOW";
18
19#[cfg(test)]
20thread_local! {
21 static REACTIVE_SHADOW_DELTA_FOLD_TARGET_PERTURBATION: Cell<bool> =
22 const { Cell::new(false) };
23}
24
25#[cfg(test)]
26pub(crate) fn set_reactive_shadow_delta_fold_target_perturbation_for_test(enabled: bool) {
27 REACTIVE_SHADOW_DELTA_FOLD_TARGET_PERTURBATION.with(|cell| cell.set(enabled));
28}
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
31pub(crate) enum ReactiveShadowPublishTierV0 {
32 Baseline,
33 Optimizing,
34}
35
36#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37pub(crate) struct ReactiveShadowStampsV0 {
38 pub(crate) corpus_revision: u64,
39 pub(crate) style_snapshot_revision: u64,
40 pub(crate) demand_generation: u64,
41}
42
43impl ReactiveShadowStampsV0 {
44 pub(crate) fn as_state(self) -> ReactiveStateV0 {
45 ReactiveStateV0::available(ReactiveValueV0::Tuple(vec![
46 ReactiveValueV0::Counter(self.corpus_revision),
47 ReactiveValueV0::Counter(self.style_snapshot_revision),
48 ReactiveValueV0::Counter(self.demand_generation),
49 ]))
50 }
51}
52
53#[derive(Debug, Clone, PartialEq, Eq)]
54pub(crate) struct ReactiveShadowDeliveryDecisionV0 {
55 pub(crate) candidate_id: u64,
56 pub(crate) uri: String,
57 pub(crate) tier: Option<ReactiveShadowPublishTierV0>,
58 pub(crate) should_deliver: bool,
59}
60
61#[derive(Debug, Clone, PartialEq, Eq)]
62pub(crate) struct ReactiveShadowFlushReportV0 {
63 pub(crate) flush_id: u64,
64 pub(crate) expected_target_uris: BTreeSet<String>,
65 pub(crate) projected_target_uris: BTreeSet<String>,
66 pub(crate) expected_stamps: ReactiveShadowStampsV0,
67 pub(crate) projected_stamps: Option<ReactiveShadowStampsV0>,
68 pub(crate) expected_baseline_digests: BTreeMap<String, String>,
69 pub(crate) projected_baseline_digests: BTreeMap<String, String>,
70 pub(crate) expected_optimizing_digests: BTreeMap<String, String>,
71 pub(crate) projected_optimizing_digests: BTreeMap<String, String>,
72 pub(crate) expected_delivery_decisions: Vec<ReactiveShadowDeliveryDecisionV0>,
73 pub(crate) projected_delivery_decisions: Vec<ReactiveShadowDeliveryDecisionV0>,
74 pub(crate) delta_fold_matches_full_rebuild: bool,
76 pub(crate) settled_without_pending_work: bool,
78 pub(crate) corpus_revision_reads: Vec<u64>,
79 pub(crate) snapshot_read_side_effect_count: u64,
80 pub(crate) stale_live_demand_count: u64,
81 pub(crate) observer_liveness_grounded: bool,
82}
83
84impl ReactiveShadowFlushReportV0 {
85 pub(crate) fn new(flush_id: u64, stamps: ReactiveShadowStampsV0) -> Self {
86 Self {
87 flush_id,
88 expected_target_uris: BTreeSet::new(),
89 projected_target_uris: BTreeSet::new(),
90 expected_stamps: stamps,
91 projected_stamps: None,
92 expected_baseline_digests: BTreeMap::new(),
93 projected_baseline_digests: BTreeMap::new(),
94 expected_optimizing_digests: BTreeMap::new(),
95 projected_optimizing_digests: BTreeMap::new(),
96 expected_delivery_decisions: Vec::new(),
97 projected_delivery_decisions: Vec::new(),
98 delta_fold_matches_full_rebuild: false,
99 settled_without_pending_work: false,
100 corpus_revision_reads: vec![stamps.corpus_revision],
101 snapshot_read_side_effect_count: 0,
102 stale_live_demand_count: 0,
103 observer_liveness_grounded: false,
104 }
105 }
106}
107
108#[derive(Debug, Clone, PartialEq, Eq)]
109struct ReactiveShadowPublishCandidateV0 {
110 candidate_id: u64,
111 flush_id: u64,
112 uri: String,
113 tier: Option<ReactiveShadowPublishTierV0>,
114 digest: Option<String>,
115 terminal_for_revision: bool,
116}
117
118#[derive(Clone)]
119pub(crate) struct ReactiveShadowPublishReceiptV0 {
120 observer: ReactiveShadowObserverV0,
121 candidate: ReactiveShadowPublishCandidateV0,
122}
123
124impl fmt::Debug for ReactiveShadowPublishReceiptV0 {
125 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
126 formatter
127 .debug_struct("ReactiveShadowPublishReceiptV0")
128 .field("candidate", &self.candidate)
129 .finish_non_exhaustive()
130 }
131}
132
133impl ReactiveShadowPublishReceiptV0 {
134 pub(crate) fn record_delivery_decision(&self, should_deliver: bool) {
135 self.observer
136 .record_delivery_decision(&self.candidate, should_deliver);
137 }
138
139 pub(crate) fn record_delivered(&self) {
140 self.observer.record_delivered(&self.candidate);
141 }
142}
143
144#[derive(Clone)]
145pub(crate) struct ReactiveShadowObserverV0 {
146 inner: Arc<Mutex<ReactiveShadowDriverV0>>,
147}
148
149impl fmt::Debug for ReactiveShadowObserverV0 {
150 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
151 formatter
152 .debug_struct("ReactiveShadowObserverV0")
153 .field("process_local", &true)
154 .finish_non_exhaustive()
155 }
156}
157
158impl ReactiveShadowObserverV0 {
159 pub(crate) fn new() -> Result<Self, String> {
160 Ok(Self {
161 inner: Arc::new(Mutex::new(ReactiveShadowDriverV0::new()?)),
162 })
163 }
164
165 pub(crate) fn begin_flush(&self, stamps: ReactiveShadowStampsV0) -> Option<u64> {
166 self.with_driver(|driver| driver.begin_flush(stamps))
167 .flatten()
168 }
169
170 pub(crate) fn current_flush_id(&self) -> Option<u64> {
171 self.with_driver(|driver| driver.active_flush_id).flatten()
172 }
173
174 pub(crate) fn record_tier_digest(
175 &self,
176 flush_id: Option<u64>,
177 uri: &str,
178 tier: ReactiveShadowPublishTierV0,
179 digest: &str,
180 terminal_for_revision: bool,
181 ) -> Option<ReactiveShadowPublishReceiptV0> {
182 self.with_driver(|driver| {
183 driver.record_tier_digest(flush_id?, uri, tier, digest, terminal_for_revision)
184 })
185 .flatten()
186 .map(|candidate| ReactiveShadowPublishReceiptV0 {
187 observer: self.clone(),
188 candidate,
189 })
190 }
191
192 pub(crate) fn record_clear(
193 &self,
194 flush_id: Option<u64>,
195 uri: &str,
196 ) -> Option<ReactiveShadowPublishReceiptV0> {
197 self.with_driver(|driver| driver.record_clear(flush_id?, uri))
198 .flatten()
199 .map(|candidate| ReactiveShadowPublishReceiptV0 {
200 observer: self.clone(),
201 candidate,
202 })
203 }
204
205 pub(crate) fn complete_flush(
206 &self,
207 flush_id: u64,
208 expected_target_uris: BTreeSet<String>,
209 independently_projected_target_uris: BTreeSet<String>,
210 final_stamps: ReactiveShadowStampsV0,
211 ) {
212 let _ = self.with_driver(|driver| {
213 driver.complete_flush(
214 flush_id,
215 expected_target_uris,
216 independently_projected_target_uris,
217 final_stamps,
218 );
219 });
220 }
221
222 pub(crate) fn module_interface_changed(
223 &self,
224 uri: &str,
225 projection: Option<omena_query::OmenaQueryModuleInterfaceChangeProjectionV0>,
226 ) -> bool {
227 self.with_driver(|driver| driver.module_interface_changed(uri, projection))
228 .unwrap_or(true)
229 }
230
231 pub(crate) fn forget_module_interface(&self, uri: &str) {
232 let _ = self.with_driver(|driver| {
233 driver.module_interface_projections.remove(uri);
234 });
235 }
236
237 #[cfg(test)]
238 pub(crate) fn reports(&self) -> Vec<ReactiveShadowFlushReportV0> {
239 self.with_driver(|driver| driver.flushes.values().cloned().collect())
240 .unwrap_or_default()
241 }
242
243 #[cfg(test)]
244 pub(crate) fn failures(&self) -> Vec<String> {
245 self.with_driver(|driver| driver.failures.clone())
246 .unwrap_or_else(|| vec!["reactive shadow observer lock was poisoned".to_string()])
247 }
248
249 #[cfg(test)]
250 pub(crate) fn inject_snapshot_read_side_effect_for_test(&self, flush_id: u64) {
251 let _ = self.with_driver(|driver| {
252 let snapshot_state = driver.engine.state(driver.stamp_input).cloned().ok();
253 if let Some(snapshot_state) = snapshot_state {
254 driver.deposit(driver.stamp_input, snapshot_state);
255 driver.settle();
256 if let Some(flush) = driver.flushes.get_mut(&flush_id) {
257 flush.snapshot_read_side_effect_count =
258 flush.snapshot_read_side_effect_count.saturating_add(1);
259 }
260 }
261 });
262 }
263
264 fn record_delivery_decision(
265 &self,
266 candidate: &ReactiveShadowPublishCandidateV0,
267 should_deliver: bool,
268 ) {
269 let _ = self.with_driver(|driver| {
270 driver.record_delivery_decision(candidate, should_deliver);
271 });
272 }
273
274 fn record_delivered(&self, candidate: &ReactiveShadowPublishCandidateV0) {
275 let _ = self.with_driver(|driver| driver.record_delivered(candidate));
276 }
277
278 fn with_driver<T>(
279 &self,
280 operation: impl FnOnce(&mut ReactiveShadowDriverV0) -> T,
281 ) -> Option<T> {
282 let mut driver = self.inner.lock().ok()?;
283 Some(operation(&mut driver))
284 }
285}
286
287struct ReactiveShadowDriverV0 {
288 engine: ReactiveEngineV0,
289 target_set_input: ReactiveNodeIdV0,
290 target_set_projection: ReactiveNodeIdV0,
291 stamp_input: ReactiveNodeIdV0,
292 baseline_digest_input: ReactiveNodeIdV0,
293 baseline_digest_projection: ReactiveNodeIdV0,
294 optimizing_digest_input: ReactiveNodeIdV0,
295 optimizing_digest_projection: ReactiveNodeIdV0,
296 digest_fold: ReactiveNodeIdV0,
297 delivery_decision_input: ReactiveNodeIdV0,
298 delivery_effect: ReactiveNodeIdV0,
299 active_flush_id: Option<u64>,
300 next_flush_id: u64,
301 next_candidate_id: u64,
302 baseline_digests: BTreeMap<String, String>,
303 optimizing_digests: BTreeMap<String, String>,
304 delivered_by_tier: BTreeMap<(String, ReactiveShadowPublishTierV0), String>,
305 delivered_current_by_uri: BTreeMap<String, String>,
306 module_interface_projections:
307 BTreeMap<String, omena_query::OmenaQueryModuleInterfaceChangeProjectionV0>,
308 latest_flush_by_uri: BTreeMap<String, u64>,
309 flushes: BTreeMap<u64, ReactiveShadowFlushReportV0>,
310 failures: Vec<String>,
311}
312
313impl ReactiveShadowDriverV0 {
314 fn new() -> Result<Self, String> {
315 let mut graph = ReactiveGraphBuilderV0::new();
316 let target_set_input = graph.add_input(
317 string_set_state(BTreeSet::new()),
318 ChangePolicyV0::exact("affectedTargetSetDeposit"),
319 );
320 let target_set_projection = graph.add_map(
321 target_set_input,
322 clone_state,
323 ChangePolicyV0::exact("affectedTargetSetProjection"),
324 );
325 let stamp_input = graph.add_input(
326 ReactiveShadowStampsV0 {
327 corpus_revision: 0,
328 style_snapshot_revision: 0,
329 demand_generation: 0,
330 }
331 .as_state(),
332 ChangePolicyV0::exact("snapshotGenerationDeposit"),
333 );
334 let baseline_digest_input = graph.add_input(
335 text_map_state(BTreeMap::new()),
336 ChangePolicyV0::exact("baselineDigestDeposit"),
337 );
338 let baseline_digest_projection = graph.add_map(
339 baseline_digest_input,
340 clone_state,
341 ChangePolicyV0::exact("baselineDigestProjection"),
342 );
343 let optimizing_digest_input = graph.add_input(
344 text_map_state(BTreeMap::new()),
345 ChangePolicyV0::exact("optimizingDigestDeposit"),
346 );
347 let optimizing_digest_projection = graph.add_map(
348 optimizing_digest_input,
349 clone_state,
350 ChangePolicyV0::exact("optimizingDigestProjection"),
351 );
352 let digest_fold = graph
353 .add_delta_fold(
354 vec![
355 ("baseline".to_string(), baseline_digest_projection),
356 ("optimizing".to_string(), optimizing_digest_projection),
357 ],
358 ChangePolicyV0::exact("tierDigestFold"),
359 )
360 .map_err(|error| error.to_string())?;
361 let delivery_decision_input = graph.add_input(
362 delivery_state(0, false),
363 ChangePolicyV0::exact("deliveryDecisionDeposit"),
364 );
365 let delivery_effect = graph.add_effect_boundary(
366 delivery_decision_input,
367 DELIVERY_EFFECT_CHANNEL,
368 ChangePolicyV0::exact("deliveryDecisionReceipt"),
369 );
370 let mut engine = graph.build().map_err(|error| error.to_string())?;
371 for node in [
372 target_set_projection,
373 stamp_input,
374 digest_fold,
375 delivery_effect,
376 ] {
377 engine.observe(node).map_err(|error| error.to_string())?;
378 }
379 let _ = engine
380 .stabilize_until_settled(STABILIZATION_RECOMPUTE_LIMIT)
381 .map_err(|error| error.to_string())?;
382 let _ = engine.drain_effect_receipts();
383
384 Ok(Self {
385 engine,
386 target_set_input,
387 target_set_projection,
388 stamp_input,
389 baseline_digest_input,
390 baseline_digest_projection,
391 optimizing_digest_input,
392 optimizing_digest_projection,
393 digest_fold,
394 delivery_decision_input,
395 delivery_effect,
396 active_flush_id: None,
397 next_flush_id: 0,
398 next_candidate_id: 0,
399 baseline_digests: BTreeMap::new(),
400 optimizing_digests: BTreeMap::new(),
401 delivered_by_tier: BTreeMap::new(),
402 delivered_current_by_uri: BTreeMap::new(),
403 module_interface_projections: BTreeMap::new(),
404 latest_flush_by_uri: BTreeMap::new(),
405 flushes: BTreeMap::new(),
406 failures: Vec::new(),
407 })
408 }
409
410 fn begin_flush(&mut self, stamps: ReactiveShadowStampsV0) -> Option<u64> {
411 if let Some(active_flush_id) = self.active_flush_id {
412 self.failures.push(format!(
413 "flush {active_flush_id} was still active when another flush began"
414 ));
415 return None;
416 }
417 self.next_flush_id = self.next_flush_id.saturating_add(1).max(1);
418 let flush_id = self.next_flush_id;
419 self.flushes
420 .insert(flush_id, ReactiveShadowFlushReportV0::new(flush_id, stamps));
421 self.active_flush_id = Some(flush_id);
422 Some(flush_id)
423 }
424
425 fn record_tier_digest(
426 &mut self,
427 flush_id: u64,
428 uri: &str,
429 tier: ReactiveShadowPublishTierV0,
430 digest: &str,
431 terminal_for_revision: bool,
432 ) -> Option<ReactiveShadowPublishCandidateV0> {
433 if !self.flushes.contains_key(&flush_id) {
434 self.failures
435 .push(format!("tier digest referenced unknown flush {flush_id}"));
436 return None;
437 }
438 match tier {
439 ReactiveShadowPublishTierV0::Baseline => {
440 self.baseline_digests
441 .insert(uri.to_string(), digest.to_string());
442 }
443 ReactiveShadowPublishTierV0::Optimizing => {
444 self.optimizing_digests
445 .insert(uri.to_string(), digest.to_string());
446 }
447 }
448 self.next_candidate_id = self.next_candidate_id.saturating_add(1).max(1);
449 let candidate = ReactiveShadowPublishCandidateV0 {
450 candidate_id: self.next_candidate_id,
451 flush_id,
452 uri: uri.to_string(),
453 tier: Some(tier),
454 digest: Some(digest.to_string()),
455 terminal_for_revision,
456 };
457 if self.active_flush_id != Some(flush_id) {
458 self.sync_digest_projection(flush_id);
459 }
460 Some(candidate)
461 }
462
463 fn record_clear(
464 &mut self,
465 flush_id: u64,
466 uri: &str,
467 ) -> Option<ReactiveShadowPublishCandidateV0> {
468 if !self.flushes.contains_key(&flush_id) {
469 self.failures
470 .push(format!("clear referenced unknown flush {flush_id}"));
471 return None;
472 }
473 self.baseline_digests.remove(uri);
474 self.optimizing_digests.remove(uri);
475 self.next_candidate_id = self.next_candidate_id.saturating_add(1).max(1);
476 Some(ReactiveShadowPublishCandidateV0 {
477 candidate_id: self.next_candidate_id,
478 flush_id,
479 uri: uri.to_string(),
480 tier: None,
481 digest: None,
482 terminal_for_revision: true,
483 })
484 }
485
486 fn complete_flush(
487 &mut self,
488 flush_id: u64,
489 expected_target_uris: BTreeSet<String>,
490 independently_projected_target_uris: BTreeSet<String>,
491 final_stamps: ReactiveShadowStampsV0,
492 ) {
493 if self.active_flush_id != Some(flush_id) {
494 self.failures
495 .push(format!("flush {flush_id} completed out of order"));
496 return;
497 }
498 if !self.flushes.contains_key(&flush_id) {
499 self.failures
500 .push(format!("flush {flush_id} completed without a record"));
501 self.active_flush_id = None;
502 return;
503 }
504 self.deposit(
505 self.target_set_input,
506 string_set_state(independently_projected_target_uris),
507 );
508 self.deposit(self.stamp_input, final_stamps.as_state());
509 self.deposit(
510 self.baseline_digest_input,
511 text_map_state(self.baseline_digests.clone()),
512 );
513 self.deposit(
514 self.optimizing_digest_input,
515 text_map_state(self.optimizing_digests.clone()),
516 );
517 self.settle();
518
519 let projected_target_uris =
520 string_set_from_state(self.engine.state(self.target_set_projection).ok());
521 let projected_stamps = stamps_from_state(self.engine.state(self.stamp_input).ok());
522 let projected_baseline_digests =
523 text_map_from_state(self.engine.state(self.baseline_digest_projection).ok());
524 let projected_optimizing_digests =
525 text_map_from_state(self.engine.state(self.optimizing_digest_projection).ok());
526 let delta_fold_matches_full_rebuild = self.delta_fold_matches_full_rebuild();
527 let settled_without_pending_work = !self.engine.has_pending_work();
528 let observer_liveness_grounded = [
529 self.target_set_projection,
530 self.stamp_input,
531 self.digest_fold,
532 self.delivery_effect,
533 ]
534 .into_iter()
535 .all(|node| self.engine.is_necessary(node).unwrap_or(false));
536
537 for uri in &expected_target_uris {
538 self.latest_flush_by_uri.insert(uri.clone(), flush_id);
539 }
540 if let Some(flush) = self.flushes.get_mut(&flush_id) {
541 flush.expected_target_uris = expected_target_uris;
542 flush.projected_target_uris = projected_target_uris;
543 flush.expected_stamps = final_stamps;
544 flush.projected_stamps = projected_stamps;
545 flush.expected_baseline_digests = self.baseline_digests.clone();
546 flush.projected_baseline_digests = projected_baseline_digests;
547 flush.expected_optimizing_digests = self.optimizing_digests.clone();
548 flush.projected_optimizing_digests = projected_optimizing_digests;
549 flush.delta_fold_matches_full_rebuild = delta_fold_matches_full_rebuild;
550 flush.settled_without_pending_work = settled_without_pending_work;
551 flush
552 .corpus_revision_reads
553 .push(final_stamps.corpus_revision);
554 flush.observer_liveness_grounded = observer_liveness_grounded;
555 }
556 self.active_flush_id = None;
557 }
558
559 fn record_delivery_decision(
560 &mut self,
561 candidate: &ReactiveShadowPublishCandidateV0,
562 should_deliver: bool,
563 ) {
564 let expected_decision = ReactiveShadowDeliveryDecisionV0 {
565 candidate_id: candidate.candidate_id,
566 uri: candidate.uri.clone(),
567 tier: candidate.tier,
568 should_deliver,
569 };
570 let projected_should_deliver = self.project_delivery_decision(candidate);
571 let projected_decision = ReactiveShadowDeliveryDecisionV0 {
572 should_deliver: projected_should_deliver,
573 ..expected_decision.clone()
574 };
575 let Some(flush) = self.flushes.get_mut(&candidate.flush_id) else {
576 self.failures.push(format!(
577 "delivery decision referenced unknown flush {}",
578 candidate.flush_id
579 ));
580 return;
581 };
582 if flush
583 .expected_delivery_decisions
584 .iter()
585 .any(|existing| existing.candidate_id == candidate.candidate_id)
586 {
587 return;
588 }
589 if self
590 .latest_flush_by_uri
591 .get(candidate.uri.as_str())
592 .is_some_and(|latest_flush_id| *latest_flush_id != candidate.flush_id)
593 && should_deliver
594 {
595 flush.stale_live_demand_count = flush.stale_live_demand_count.saturating_add(1);
596 }
597 flush.expected_delivery_decisions.push(expected_decision);
598 self.deposit(
599 self.delivery_decision_input,
600 delivery_state(candidate.candidate_id, projected_should_deliver),
601 );
602 self.settle();
603 let receipts = self.engine.drain_effect_receipts();
604 let observed = receipts.iter().any(|receipt| {
605 receipt.channel == DELIVERY_EFFECT_CHANNEL
606 && receipt.state == delivery_state(candidate.candidate_id, projected_should_deliver)
607 });
608 if observed {
609 if let Some(flush) = self.flushes.get_mut(&candidate.flush_id) {
610 flush.projected_delivery_decisions.push(projected_decision);
611 flush.settled_without_pending_work = !self.engine.has_pending_work();
612 }
613 } else {
614 self.failures.push(format!(
615 "delivery decision {} produced no effect receipt",
616 candidate.candidate_id
617 ));
618 }
619 }
620
621 fn project_delivery_decision(&self, candidate: &ReactiveShadowPublishCandidateV0) -> bool {
622 let (Some(tier), Some(digest)) = (candidate.tier, candidate.digest.as_ref()) else {
623 return true;
624 };
625 if self.delivered_by_tier.get(&(candidate.uri.clone(), tier)) != Some(digest) {
626 return true;
627 }
628 candidate.terminal_for_revision
629 && self.delivered_current_by_uri.get(candidate.uri.as_str()) != Some(digest)
630 }
631
632 fn record_delivered(&mut self, candidate: &ReactiveShadowPublishCandidateV0) {
633 match (candidate.tier, candidate.digest.as_ref()) {
634 (Some(tier), Some(digest)) => {
635 self.delivered_by_tier
636 .insert((candidate.uri.clone(), tier), digest.clone());
637 self.delivered_current_by_uri
638 .insert(candidate.uri.clone(), digest.clone());
639 }
640 _ => {
641 self.delivered_by_tier
642 .retain(|(uri, _), _| uri != candidate.uri.as_str());
643 self.delivered_current_by_uri.remove(candidate.uri.as_str());
644 }
645 }
646 }
647
648 fn module_interface_changed(
649 &mut self,
650 uri: &str,
651 projection: Option<omena_query::OmenaQueryModuleInterfaceChangeProjectionV0>,
652 ) -> bool {
653 let Some(projection) = projection else {
654 return true;
655 };
656 if self
657 .module_interface_projections
658 .get(uri)
659 .is_some_and(|previous| *previous == projection)
660 {
661 return false;
662 }
663 self.module_interface_projections
664 .insert(uri.to_string(), projection);
665 while self.module_interface_projections.len() > MODULE_INTERFACE_MEMO_ENTRY_LIMIT {
666 let evicted_uri = self
667 .module_interface_projections
668 .keys()
669 .find(|candidate| candidate.as_str() != uri)
670 .cloned()
671 .or_else(|| self.module_interface_projections.keys().next().cloned());
672 let Some(evicted_uri) = evicted_uri else {
673 break;
674 };
675 self.module_interface_projections
676 .remove(evicted_uri.as_str());
677 }
678 true
679 }
680
681 fn sync_digest_projection(&mut self, flush_id: u64) {
682 self.deposit(
683 self.baseline_digest_input,
684 text_map_state(self.baseline_digests.clone()),
685 );
686 self.deposit(
687 self.optimizing_digest_input,
688 text_map_state(self.optimizing_digests.clone()),
689 );
690 self.settle();
691 let projected_baseline_digests =
692 text_map_from_state(self.engine.state(self.baseline_digest_projection).ok());
693 let projected_optimizing_digests =
694 text_map_from_state(self.engine.state(self.optimizing_digest_projection).ok());
695 let delta_fold_matches_full_rebuild = self.delta_fold_matches_full_rebuild();
696 if let Some(flush) = self.flushes.get_mut(&flush_id) {
697 flush.expected_baseline_digests = self.baseline_digests.clone();
698 flush.projected_baseline_digests = projected_baseline_digests;
699 flush.expected_optimizing_digests = self.optimizing_digests.clone();
700 flush.projected_optimizing_digests = projected_optimizing_digests;
701 flush.delta_fold_matches_full_rebuild = delta_fold_matches_full_rebuild;
702 flush.settled_without_pending_work = !self.engine.has_pending_work();
703 }
704 }
705
706 fn deposit(&mut self, node: ReactiveNodeIdV0, state: ReactiveStateV0) {
707 if let Err(error) = self.engine.deposit(node, state) {
708 self.failures.push(error.to_string());
709 }
710 }
711
712 fn delta_fold_matches_full_rebuild(&self) -> bool {
713 #[cfg(test)]
714 let node = if REACTIVE_SHADOW_DELTA_FOLD_TARGET_PERTURBATION.with(Cell::get) {
715 self.baseline_digest_projection
716 } else {
717 self.digest_fold
718 };
719 #[cfg(not(test))]
720 let node = self.digest_fold;
721 self.engine.verify_delta_fold(node).is_ok()
722 }
723
724 fn settle(&mut self) {
725 match self
726 .engine
727 .stabilize_until_settled(STABILIZATION_RECOMPUTE_LIMIT)
728 {
729 Ok(StabilizeStatusV0::Settled { .. }) => {}
730 Ok(StabilizeStatusV0::Pending { .. }) => self
731 .failures
732 .push("reactive shadow exceeded its bounded stabilization budget".to_string()),
733 Ok(_) => self
734 .failures
735 .push("reactive shadow observed an unknown stabilization status".to_string()),
736 Err(error) => self.failures.push(error.to_string()),
737 }
738 }
739}
740
741impl crate::LspShellState {
742 pub fn enable_reactive_shadow_observer(&mut self) -> Result<(), String> {
745 self.diagnostics_publish_digest_registry
746 .enable_reactive_shadow()
747 }
748}
749
750fn clone_state(state: &ReactiveStateV0) -> ReactiveStateV0 {
751 state.clone()
752}
753
754fn string_set_state(values: BTreeSet<String>) -> ReactiveStateV0 {
755 ReactiveStateV0::available(ReactiveValueV0::StringSet(values))
756}
757
758fn text_map_state(values: BTreeMap<String, String>) -> ReactiveStateV0 {
759 ReactiveStateV0::available(ReactiveValueV0::TextMap(values))
760}
761
762fn delivery_state(candidate_id: u64, should_deliver: bool) -> ReactiveStateV0 {
763 ReactiveStateV0::available(ReactiveValueV0::Tuple(vec![
764 ReactiveValueV0::Counter(candidate_id),
765 ReactiveValueV0::Bool(should_deliver),
766 ]))
767}
768
769fn string_set_from_state(state: Option<&ReactiveStateV0>) -> BTreeSet<String> {
770 match state {
771 Some(ReactiveStateV0::Available(ReactiveValueV0::StringSet(values))) => values.clone(),
772 _ => BTreeSet::new(),
773 }
774}
775
776fn text_map_from_state(state: Option<&ReactiveStateV0>) -> BTreeMap<String, String> {
777 match state {
778 Some(ReactiveStateV0::Available(ReactiveValueV0::TextMap(values))) => values.clone(),
779 _ => BTreeMap::new(),
780 }
781}
782
783pub(crate) fn stamps_from_state(state: Option<&ReactiveStateV0>) -> Option<ReactiveShadowStampsV0> {
784 let Some(ReactiveStateV0::Available(ReactiveValueV0::Tuple(values))) = state else {
785 return None;
786 };
787 let [
788 ReactiveValueV0::Counter(corpus_revision),
789 ReactiveValueV0::Counter(style_snapshot_revision),
790 ReactiveValueV0::Counter(demand_generation),
791 ] = values.as_slice()
792 else {
793 return None;
794 };
795 Some(ReactiveShadowStampsV0 {
796 corpus_revision: *corpus_revision,
797 style_snapshot_revision: *style_snapshot_revision,
798 demand_generation: *demand_generation,
799 })
800}