omena_lsp_server/
tide_republish.rs

1//! M3: the off-loop workspace-republish executor (rfcs#111 §8.5, §12 M3).
2//!
3//! `prepare` flushes the republish lane through its settle gate on the loop
4//! and captures a copy-on-write query snapshot; `collect` runs the
5//! abort-capable parallel wave against that snapshot on a worker thread —
6//! the loop stays free; `apply` publishes loop-side in canonical order under
7//! the loop's per-tick chunk budget, writing behind through the LOOP state's
8//! disk-cache session (the snapshot carries a default session on purpose).
9//! A settle-window reopen bumps the lane generation: the wave aborts at the
10//! next item boundary and pending applies are dropped — their keys are
11//! republished by the reopened window's tide, and the publication order key
12//! forbids any stale overwrite.
13
14use crate::LspShellState;
15use crate::diagnostics_follow_up::{
16    TIDE_REPUBLISH_LANE_CONFIG, workspace_republish_frontier_passed,
17};
18use crate::disk_cache::DiskDiagnosticsCacheSlotV0;
19use crate::lsp_output::ScheduledLspOutput;
20use crate::state::LspQuerySnapshotV0;
21use crate::tide::TideGateInputsV0;
22use serde_json::Value;
23use std::sync::Arc;
24use std::sync::atomic::{AtomicU64, Ordering};
25
26#[derive(Debug)]
27pub struct TideWorkspaceRepublishJobV0 {
28    snapshot: LspQuerySnapshotV0,
29    uris: Vec<String>,
30    pub generation: u64,
31    gen_watch: Arc<AtomicU64>,
32}
33
34impl TideWorkspaceRepublishJobV0 {
35    #[cfg(test)]
36    pub(crate) fn target_uris_for_test(&self) -> &[String] {
37        self.uris.as_slice()
38    }
39}
40
41#[derive(Debug)]
42pub struct TideWorkspaceRepublishItemV0 {
43    pub(crate) uri: String,
44    pub(crate) diagnostics: Value,
45    pub(crate) disk_cache_slot: Option<DiskDiagnosticsCacheSlotV0>,
46}
47
48#[derive(Debug)]
49pub struct TideWorkspaceRepublishResultV0 {
50    pub generation: u64,
51    pub items: Vec<TideWorkspaceRepublishItemV0>,
52    /// Wave-ineligible targets of THIS chunk. When the tide is still current
53    /// at completion these fall back to the per-file deferred arm; a
54    /// disowned tide drops them (the reopened window covers the corpus).
55    pub uncovered_uris: Vec<String>,
56    /// The last chunk of the tide: the loop completes the tide once this
57    /// arrives and the apply queue drains.
58    pub final_chunk: bool,
59}
60
61/// Gate evaluation + snapshot capture, on the loop. `idle` is the courtesy
62/// input (no recent client message); aging overrides it after
63/// [`TIDE_REPUBLISH_LANE_CONFIG`]'s bound, the frontier never.
64pub fn prepare_tide_workspace_republish_job(
65    state: &mut LspShellState,
66    idle: bool,
67) -> Option<TideWorkspaceRepublishJobV0> {
68    if !state.external_sif_refresh_deferred {
69        return None;
70    }
71    let inputs = TideGateInputsV0 {
72        frontier_passed: workspace_republish_frontier_passed(state),
73        idle,
74    };
75    let flush = state.tide_republish_lane.try_flush(
76        inputs,
77        state.tide_tick,
78        &TIDE_REPUBLISH_LANE_CONFIG,
79    )?;
80    state
81        .tide_republish_gen_watch
82        .store(flush.generation, Ordering::Relaxed);
83    // Demand-shaped targeting (rfcs#111 demand lattice): `All` covers the
84    // corpus, `Cone` covers the seeds' reverse-dependency closure at flush
85    // time. Open documents come first — the user is looking at them — and
86    // chunked streaming below turns that ordering into convergence latency.
87    let uris = crate::diagnostics_follow_up::tide_republish_target_uris(state, &flush.demand);
88    if uris.is_empty() {
89        state.tide_republish_lane.tide_completed(flush.generation);
90        return None;
91    }
92    crate::loop_trace!(
93        "republish-tide prepared gen={} targets={} demand={}",
94        flush.generation,
95        uris.len(),
96        match &flush.demand {
97            crate::tide::TideRepublishDemandV0::All => "all".to_string(),
98            crate::tide::TideRepublishDemandV0::Cone(seeds) => format!("cone({})", seeds.len()),
99            crate::tide::TideRepublishDemandV0::None => "none".to_string(),
100        }
101    );
102    Some(TideWorkspaceRepublishJobV0 {
103        snapshot: state.query_snapshot(),
104        uris,
105        generation: flush.generation,
106        gen_watch: Arc::clone(&state.tide_republish_gen_watch),
107    })
108}
109
110/// Worker-side compute, streaming (rfcs#111 §8.5): ONE shared-graph parallel
111/// wave — one memo-host sync, one substrate, one condensation — with a
112/// per-item sink that emits each target the moment its pool task finishes.
113/// Open documents were ordered first by prepare, so they converge first.
114/// The generation watch aborts disowned tides at item boundaries; the final
115/// event carries the uncovered remainder for the fallback arm.
116pub fn collect_tide_workspace_republish_streaming(
117    job: TideWorkspaceRepublishJobV0,
118    emit: &(dyn Fn(TideWorkspaceRepublishResultV0) -> bool + Sync),
119) {
120    // A panic in the shared setup (host sync, substrate, condensation)
121    // would otherwise unwind past the final-chunk emit, leaving the loop's
122    // in-flight gate stuck at 1 and workspace republish silently dead for
123    // the session. Guarantee a final chunk on EVERY exit: on panic, report
124    // every target uncovered — the fallback arm re-covers them per-file,
125    // and items already streamed are superseded harmlessly.
126    let generation = job.generation;
127    let uncovered_fallback = job.uris.clone();
128    if std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
129        collect_tide_workspace_republish_streaming_inner(job, emit)
130    }))
131    .is_err()
132    {
133        let _ = emit(TideWorkspaceRepublishResultV0 {
134            generation,
135            items: Vec::new(),
136            uncovered_uris: uncovered_fallback,
137            final_chunk: true,
138        });
139    }
140}
141
142fn collect_tide_workspace_republish_streaming_inner(
143    job: TideWorkspaceRepublishJobV0,
144    emit: &(dyn Fn(TideWorkspaceRepublishResultV0) -> bool + Sync),
145) {
146    let covered = std::sync::Mutex::new(std::collections::BTreeSet::<usize>::new());
147    let sink =
148        |index: usize,
149         diagnostics: serde_json::Value,
150         disk_cache_slot: Option<crate::disk_cache::DiskDiagnosticsCacheSlotV0>| {
151            if let Ok(mut covered) = covered.lock() {
152                covered.insert(index);
153            }
154            let Some(uri) = job.uris.get(index) else {
155                return;
156            };
157            let _ = emit(TideWorkspaceRepublishResultV0 {
158                generation: job.generation,
159                items: vec![TideWorkspaceRepublishItemV0 {
160                    uri: uri.clone(),
161                    diagnostics,
162                    disk_cache_slot,
163                }],
164                uncovered_uris: Vec::new(),
165                final_chunk: false,
166            });
167        };
168    let _ = crate::parallel_style_wave::resolved_parallel_style_wave_targets_from_read_view_with_abort_and_sink(
169        &job.snapshot,
170        job.uris.as_slice(),
171        crate::parallel_style_wave::PARALLEL_STYLE_WAVE_MIN_PARALLEL_TARGETS,
172        Some((job.gen_watch.as_ref(), job.generation)),
173        Some(&sink),
174    );
175    let covered = covered.into_inner().unwrap_or_default();
176    let uncovered_uris = job
177        .uris
178        .iter()
179        .enumerate()
180        .filter(|(index, _)| !covered.contains(index))
181        .map(|(_, uri)| uri.clone())
182        .collect::<Vec<_>>();
183    crate::loop_trace!(
184        "republish-tide collected gen={} covered={} uncovered={}",
185        job.generation,
186        covered.len(),
187        uncovered_uris.len()
188    );
189    let _ = emit(TideWorkspaceRepublishResultV0 {
190        generation: job.generation,
191        items: Vec::new(),
192        uncovered_uris,
193        final_chunk: true,
194    });
195}
196
197/// Loop-side apply for ONE item — the caller pumps a bounded chunk per tick
198/// (I4) and must have verified the tide generation is still current.
199/// Write-behind runs through the loop state's real disk-cache session, then
200/// the tiered publish emits in the same shape as every other arm.
201pub fn apply_tide_workspace_republish_item(
202    state: &mut LspShellState,
203    item: TideWorkspaceRepublishItemV0,
204) -> Vec<ScheduledLspOutput> {
205    if let Some(slot) = item.disk_cache_slot.as_ref() {
206        slot.store_write_behind(state, &item.diagnostics);
207    }
208    crate::diagnostics_scheduler::publish_tiered_diagnostics_notifications(
209        state,
210        item.uri.as_str(),
211        item.diagnostics,
212    )
213}
214
215/// Completion: re-arm the lane; when the tide is still current, uncovered
216/// targets re-enter the per-file deferred arm so no key is silently skipped.
217pub fn complete_tide_workspace_republish(
218    state: &mut LspShellState,
219    generation: u64,
220    uncovered_uris: Vec<String>,
221) -> crate::LspDiagnosticsFollowUpEffectsV0 {
222    let current = state.tide_republish_lane.generation() == generation;
223    // Read the flushed demand BEFORE discharging it: the source refresh
224    // below is shaped by the same demand that shaped the tide's own
225    // targets (review finding: an unshaped all-open-sources broadcast at
226    // every completion contradicted the cone discipline and put
227    // corpus-scale gather work on the loop).
228    let source_scope_seeds = match state.tide_republish_lane.in_flight_demand() {
229        Some(crate::tide::TideRepublishDemandV0::Cone(seeds)) => Some(seeds.clone()),
230        _ => None,
231    };
232    state.tide_republish_lane.tide_completed(generation);
233    if !current {
234        return crate::LspDiagnosticsFollowUpEffectsV0::default();
235    }
236    let mut effects = crate::diagnostics_scheduler::DiagnosticsScheduleEffectsV0::default();
237    if !uncovered_uris.is_empty() {
238        crate::loop_trace!(
239            "republish-tide leftovers gen={} n={}",
240            generation,
241            uncovered_uris.len()
242        );
243        effects.extend(
244            crate::diagnostics_scheduler::run_diagnostics_schedule_effects(
245                state,
246                crate::diagnostics_scheduler::DiagnosticsScheduleEvent::WatchedFiles {
247                    uris: uncovered_uris,
248                },
249            ),
250        );
251    }
252    // The wave re-rendered every STYLE target against the settled corpus;
253    // open SOURCE documents were rendered against whatever corpus existed
254    // when their tab opened — possibly BEFORE this settle admitted the
255    // stylesheets they reference, and nothing else ever re-evaluates them
256    // (the fan-outs fire on style EVENTS, not on corpus growth). Completion
257    // of a current-generation tide is the moment the corpus became
258    // authoritative, so open sources re-enter the same per-file deferred
259    // arm as the uncovered style leftovers.
260    let open_source_uris = crate::diagnostics_scheduler::open_document_uris_for_diagnostics(state)
261        .into_iter()
262        .filter(|uri| !crate::protocol::is_style_document_uri(uri.as_str()))
263        .collect::<Vec<_>>();
264    // Cone tides refresh only the sources that DEPEND on the cone's seeds
265    // (the same filter the didChange fan-out uses); the unscoped arm is
266    // reserved for All tides — the cold-settle case whose whole point is
267    // that every earlier source render predates the corpus.
268    let open_source_uris = match source_scope_seeds {
269        Some(seeds) => {
270            let mut scoped = std::collections::BTreeSet::new();
271            for seed in seeds {
272                scoped.extend(
273                    crate::diagnostics_scheduler::scoped_source_republish_uris_for_style_change(
274                        state,
275                        seed.as_str(),
276                        open_source_uris.clone(),
277                    ),
278                );
279            }
280            scoped.into_iter().collect::<Vec<_>>()
281        }
282        None => open_source_uris,
283    };
284    if !open_source_uris.is_empty() {
285        crate::loop_trace!(
286            "republish-tide source refresh gen={} n={}",
287            generation,
288            open_source_uris.len()
289        );
290        effects.extend(
291            crate::diagnostics_scheduler::diagnostics_effects_for_document_uris(
292                state,
293                open_source_uris,
294                true,
295            ),
296        );
297    }
298    crate::LspDiagnosticsFollowUpEffectsV0 {
299        outputs: effects.outputs,
300        deferred_diagnostics: effects.deferred_diagnostics,
301    }
302}
omena_lsp_server/tide_republish.rs

omena_lsp_server/
tide_republish.rs
