laburnum 1.17.3

// Copyright Two Neutron Stars Incorporated and contributors
// SPDX-License-Identifier: BlueOak-1.0.0

//! Observational test for GLD-0035 reactive re-dispatch across an *absent*
//! prefix dependency.
//!
//! Scenario: a watcher on partition `P1` runs a handler that queries `P2` by a
//! sort-key prefix (`BeginsWith`). We compare two handler variants:
//!
//! * Variant A re-arms a `pending_deps` re-dispatch (via
//!   `register_pending_redispatch`, exactly as `spawn_watcher_task` does after
//!   draining `defer_until`) ONLY when the `P2` prefix query is empty.
//! * Variant B re-arms unconditionally on every run.
//!
//! After the system quiesces we write NEW `P2` rows under the same prefix from
//! a *different* originating write (P1 is never touched again) and observe
//! whether the watcher re-runs.
//!
//! Harness note / deviation from the "real" path: laburnum's generic
//! `watchers!` dispatch (`T::dispatch_watcher`) is only wired for the built-in
//! Diagnostics / WorkDoneProgress partitions, so a fully macro-registered
//! user watcher over an arbitrary test partition cannot be spun up without
//! implementing the entire `LanguageServer` service surface. This test instead
//! reconstructs the watcher body faithfully: it runs as a real scheduler task
//! with a real `ctx.query_client()` prefix query, and re-arms re-dispatch
//! through the same `register_pending_redispatch` /
//! `fire_pending_redispatch` machinery (`scheduler/mod.rs:526`) that
//! `spawn_watcher_task` uses. The first run is triggered by a real `P1` commit
//! routed through `on_new_chunk`; the post-quiescence `P2` write is a real
//! commit from a different writer id, also routed through `on_new_chunk`.

use {
  crate::{
    Ident,
    database::{
      PartitionKey,
      chunk::RecordWriter,
      partitions::SortKeyOf,
      query::SortKeyCondition,
      tests::storage::{
        Test1Partition, Test2Partition, TestPartitions, TestRecordData,
      },
    },
    scheduler::{Scheduler, lanes::DEFAULT_LANE},
    source::cache::reporter::SourceCacheReader,
  },
  macro_rules_attribute::apply,
  parking_lot::Mutex,
  std::{
    sync::{
      Arc,
      atomic::{AtomicUsize, Ordering},
    },
    time::Duration,
  },
};

// P1 is the watched partition; P2 holds the prefix-queried dependency.
type P1 = Test1Partition;
type P2 = Test2Partition;

const PREFIX: &str = "dep:";

fn module_record(tag: &str) -> TestRecordData {
  TestRecordData::Module {
    exports: vec![Ident::new(tag)],
  }
}

fn new_scheduler()
-> Arc<Scheduler<TestPartitions, crate::server::LaburnumLanguageServer>> {
  let (server_conn, _client_conn) = crate::connect::ipc::Connection::memory();
  let filesystems = Arc::new(parking_lot::RwLock::new(Vec::new()));
  let source_cache =
    Arc::new(parking_lot::RwLock::new(crate::SourceCache::new()));
  let config = crate::scheduler::SchedulerConfiguration {
    rpc_response_capacity: 100,
    enable_periodic_gc: false,
    idle_debounce: Duration::from_millis(10),
  };
  let scheduler = Scheduler::new_with_config(
    server_conn,
    Arc::new(crate::server::LaburnumLanguageServer),
    filesystems,
    source_cache,
    1,
    config,
  );
  scheduler.spawn_workers();
  scheduler
}

async fn wait_for<F: Fn() -> bool>(condition: F, timeout: Duration) -> bool {
  let start = std::time::Instant::now();
  while start.elapsed() < timeout {
    if condition() {
      return true;
    }
    futures_lite::future::yield_now().await;
  }
  false
}

/// Commit a single P1 record from `writer_id`, then drive the scheduler's
/// post-commit dispatch exactly as the worker loop does after a task commits.
fn commit_p1(
  scheduler: &Arc<
    Scheduler<TestPartitions, crate::server::LaburnumLanguageServer>,
  >,
  writer_id: &str,
  sort_key: &str,
  record: TestRecordData,
) {
  let mut writer = RecordWriter::new(Ident::new(writer_id));
  writer.insert::<P1>(sort_key.to_string(), record);
  let result = scheduler
    .db
    .commit_chunk(writer.build(), &SourceCacheReader::new_empty_for_test());
  scheduler.on_new_chunk(Ident::new(writer_id), result);
}

/// Like [`commit_p1`] but returns the inserted [`RecordKey`]s so a test can
/// thread the *exact* keys the commit produced into a watcher's matched-key set
/// — mirroring how `on_new_chunk` -> `spawn_watcher_task` derives `updated`.
fn commit_p1_keys(
  scheduler: &Arc<
    Scheduler<TestPartitions, crate::server::LaburnumLanguageServer>,
  >,
  writer_id: &str,
  sort_key: &str,
  record: TestRecordData,
) -> Vec<crate::database::RecordKey<TestPartitions>> {
  let mut writer = RecordWriter::new(Ident::new(writer_id));
  writer.insert::<P1>(sort_key.to_string(), record);
  let result = scheduler
    .db
    .commit_chunk(writer.build(), &SourceCacheReader::new_empty_for_test());
  let keys: Vec<_> = result.all_inserted_keys().cloned().collect();
  scheduler.on_new_chunk(Ident::new(writer_id), result);
  keys
}

/// Commit a single P2 record from `writer_id`, then drive post-commit dispatch
/// (which calls `fire_pending_redispatch`).
fn commit_p2(
  scheduler: &Arc<
    Scheduler<TestPartitions, crate::server::LaburnumLanguageServer>,
  >,
  writer_id: &str,
  sort_key: &str,
  record: TestRecordData,
) {
  let mut writer = RecordWriter::new(Ident::new(writer_id));
  writer.insert::<P2>(sort_key.to_string(), record);
  let result = scheduler
    .db
    .commit_chunk(writer.build(), &SourceCacheReader::new_empty_for_test());
  scheduler.on_new_chunk(Ident::new(writer_id), result);
}

/// Re-usable watcher body. Queues a real scheduler task that:
///   1. increments the run counter,
///   2. runs the real `P2` prefix query and records how many rows it saw,
///   3. depending on `rearm_when_empty`, re-arms a `pending_deps` re-dispatch
///      keyed by `(P2::KEY, BeginsWith(PREFIX))` — recursively re-invoking this
///      same body when a matching `P2` key later lands.
fn spawn_watcher_run(
  scheduler: Arc<
    Scheduler<TestPartitions, crate::server::LaburnumLanguageServer>,
  >,
  run_count: Arc<AtomicUsize>,
  last_seen_rows: Arc<AtomicUsize>,
  rearm_when_empty: bool,
) {
  let sched_for_task = scheduler.clone();
  scheduler.queue(
    move |mut ctx| {
      let sched = sched_for_task.clone();
      let run_count = run_count.clone();
      let last_seen_rows = last_seen_rows.clone();
      async move {
        run_count.fetch_add(1, Ordering::SeqCst);

        // Real prefix query over P2 through the task's query client.
        let results = ctx
          .query_client()
          .query(Test2Partition)
          .sort_key_begins_with(PREFIX.to_string())
          .execute()
          .await;
        let row_count = results.len();
        last_seen_rows.store(row_count, Ordering::SeqCst);

        let should_rearm = if rearm_when_empty {
          row_count == 0
        } else {
          true
        };

        if should_rearm {
          // Mirror `spawn_watcher_task`: drained `defer_until` deps become a
          // re-dispatch keyed by (partition, condition) that re-runs this body.
          let condition = SortKeyCondition::BeginsWith(<P2 as SortKeyOf<
            TestPartitions,
          >>::wrap_sort_key(
            PREFIX.to_string()
          ));
          let sched_inner = sched.clone();
          let run_count = run_count.clone();
          let last_seen_rows = last_seen_rows.clone();
          sched.register_pending_redispatch(
            crate::Ident::new("test-owner"),
            P2::KEY,
            condition,
            Box::new(move || {
              spawn_watcher_run(
                sched_inner.clone(),
                run_count.clone(),
                last_seen_rows.clone(),
                rearm_when_empty,
              );
            }),
          );
        }

        // Watcher itself writes nothing into the union here.
        None
      }
    },
    DEFAULT_LANE,
  );
}

struct Observed {
  runs_after_p1: usize,
  rows_after_p1: usize,
  runs_after_p2: usize,
  rows_after_p2: usize,
}

async fn run_variant(
  rearm_when_empty: bool,
  p2_prepopulated: bool,
) -> Observed {
  let scheduler = new_scheduler();
  let run_count = Arc::new(AtomicUsize::new(0));
  let last_seen_rows = Arc::new(AtomicUsize::new(usize::MAX));

  // Optionally seed P2 with a matching row *before* the first watcher run, so
  // the first prefix query SUCCEEDS (non-empty). This is the discriminating
  // case for "register only when empty": with the dependency already present,
  // Variant A should not re-arm.
  if p2_prepopulated {
    commit_p2(
      &scheduler,
      "p2-seed-source",
      &format!("{PREFIX}seed"),
      module_record("dep-seed"),
    );
  }

  // (a) Trigger the P1 watcher once by committing a P1 record and routing it
  // through on_new_chunk. We dispatch the watcher body ourselves (the generic
  // watcher path is not wired for arbitrary partitions — see module note),
  // then drive it to quiescence.
  commit_p1(
    &scheduler,
    "p1-source",
    "module-a",
    module_record("module-a"),
  );
  spawn_watcher_run(
    scheduler.clone(),
    run_count.clone(),
    last_seen_rows.clone(),
    rearm_when_empty,
  );

  assert!(
    wait_for(
      || run_count.load(Ordering::SeqCst) >= 1,
      Duration::from_secs(5)
    )
    .await,
    "watcher did not run after P1 trigger"
  );
  // Allow any spurious extra runs to settle.
  let _ = wait_for(|| false, Duration::from_millis(50)).await;

  let runs_after_p1 = run_count.load(Ordering::SeqCst);
  let rows_after_p1 = last_seen_rows.load(Ordering::SeqCst);

  // (b) After quiescence, a DIFFERENT source writes new P2 rows matching the
  // prefix. P1 is never touched. This routes through on_new_chunk, which calls
  // fire_pending_redispatch.
  commit_p2(
    &scheduler,
    "p2-other-source",
    &format!("{PREFIX}first"),
    module_record("dep-first"),
  );

  // (c) Quiesce again and observe.
  let baseline = runs_after_p1;
  let _ = wait_for(
    || run_count.load(Ordering::SeqCst) > baseline,
    Duration::from_secs(2),
  )
  .await;
  // Settle any cascade.
  let _ = wait_for(|| false, Duration::from_millis(100)).await;

  let runs_after_p2 = run_count.load(Ordering::SeqCst);
  let rows_after_p2 = last_seen_rows.load(Ordering::SeqCst);

  Observed {
    runs_after_p1,
    rows_after_p1,
    runs_after_p2,
    rows_after_p2,
  }
}

fn report(label: &str, o: &Observed) {
  eprintln!(
    "{label}: runs_after_p1={} rows_seen_after_p1={} runs_after_p2_write={} \
     rows_seen_on_rerun={} | re-ran on cross-source P2 write? {}",
    o.runs_after_p1,
    o.rows_after_p1,
    o.runs_after_p2,
    o.rows_after_p2,
    o.runs_after_p2 > o.runs_after_p1,
  );
}

#[apply(smol_macros::test!)]
async fn observe_reactive_prefix_redispatch_both_variants() {
  // -- Scenario 1: P2 EMPTY on first run (the "absent dependency" case) ------
  // Here both variants take the absent-dependency branch on the first run, so
  // both re-arm and both re-run.
  let a_empty = run_variant(true, false).await;
  let b_empty = run_variant(false, false).await;
  eprintln!("--- Scenario 1: P2 empty on first run (dependency absent) ---");
  report("VARIANT A (re-arm only when P2 empty)", &a_empty);
  report("VARIANT B (re-arm unconditionally)   ", &b_empty);

  // -- Scenario 2: P2 PRE-POPULATED on first run (dependency present) --------
  // The discriminating case. Variant A's first query succeeds (non-empty), so
  // it does NOT re-arm and does NOT re-run when a *new* P2 row later lands.
  // Variant B re-arms regardless and re-runs on the cross-source P2 write.
  let a_seed = run_variant(true, true).await;
  let b_seed = run_variant(false, true).await;
  eprintln!(
    "--- Scenario 2: P2 pre-populated on first run (dependency present) ---"
  );
  report("VARIANT A (re-arm only when P2 empty)", &a_seed);
  report("VARIANT B (re-arm unconditionally)   ", &b_seed);

  // -- Scenario 1 invariants (P2 empty) -------------------------------------
  assert_eq!(a_empty.rows_after_p1, 0, "P2 empty on first run");
  assert_eq!(b_empty.rows_after_p1, 0, "P2 empty on first run");
  assert!(
    a_empty.runs_after_p2 > a_empty.runs_after_p1,
    "Variant A re-runs when the first query was empty (absent-dep branch armed)"
  );
  assert!(
    b_empty.runs_after_p2 > b_empty.runs_after_p1,
    "Variant B re-runs (always armed)"
  );

  // -- Scenario 2 invariants (P2 present) -----------------------------------
  assert_eq!(
    a_seed.rows_after_p1, 1,
    "Variant A first query succeeds (sees the seeded row)"
  );
  assert_eq!(
    a_seed.runs_after_p2, a_seed.runs_after_p1,
    "Variant A does NOT re-run on the cross-source P2 write (never armed)"
  );
  assert!(
    b_seed.runs_after_p2 > b_seed.runs_after_p1,
    "Variant B re-runs on the cross-source P2 write (armed regardless)"
  );
  // On Variant B's re-run it observes the newly-added P2 row alongside the seed.
  assert!(
    b_seed.rows_after_p2 >= 2,
    "Variant B's re-run sees the new P2 row plus the seed"
  );
}

/// Per-run record of the `matched_keys` the watcher body observed via
/// `ctx.matched_keys_updated()` / `ctx.matched_keys_deleted()`.
#[derive(Default)]
struct MatchedKeyLog {
  updated_per_run: Vec<Vec<crate::database::RecordKey<TestPartitions>>>,
  deleted_per_run: Vec<Vec<crate::database::RecordKey<TestPartitions>>>,
}

/// Faithful re-creation of `spawn_watcher_task`'s replay contract
/// (`scheduler/mod.rs:866` + the re-dispatch closure at 896-916):
///
///   1. `ctx.set_matched_keys(updated.clone(), deleted.clone())` at the top of
///      EVERY invocation, then the body reads them back and logs them;
///   2. the re-dispatch thunk captures the SAME original `updated`/`deleted`,
///      so a P2-triggered re-run replays the ORIGINAL P1 keys — not the P2 key
///      that fired the re-dispatch.
///
/// Unlike `spawn_watcher_run`, this re-arms unconditionally (Variant-B style)
/// so the cross-source P2 write deterministically produces exactly one re-run.
fn spawn_watcher_run_with_keys(
  scheduler: Arc<
    Scheduler<TestPartitions, crate::server::LaburnumLanguageServer>,
  >,
  updated: Vec<crate::database::RecordKey<TestPartitions>>,
  deleted: Vec<crate::database::RecordKey<TestPartitions>>,
  run_count: Arc<AtomicUsize>,
  log: Arc<Mutex<MatchedKeyLog>>,
  rearmed: Arc<AtomicUsize>,
) {
  let sched_for_task = scheduler.clone();
  scheduler.queue(
    move |mut ctx| {
      let sched = sched_for_task.clone();
      let run_count = run_count.clone();
      let log = log.clone();
      let rearmed = rearmed.clone();
      // The original matched keys, identical for every (re-)invocation — exactly
      // what `spawn_watcher_task` clones into `set_matched_keys` and the
      // re-dispatch thunk.
      let updated = updated.clone();
      let deleted = deleted.clone();
      async move {
        // (1) Mirror spawn_watcher_task: stamp the matched keys before the body.
        ctx.set_matched_keys(updated.clone(), deleted.clone());

        run_count.fetch_add(1, Ordering::SeqCst);

        // (2) Read back what the handler actually sees this run and log it.
        let seen_updated = ctx.matched_keys_updated();
        let seen_deleted = ctx.matched_keys_deleted();
        {
          let mut log = log.lock();
          log.updated_per_run.push(seen_updated);
          log.deleted_per_run.push(seen_deleted);
        }

        // Re-arm only on the FIRST run so the cross-source P2 write yields
        // exactly one re-run (avoids an unbounded re-arm cascade).
        if rearmed.fetch_add(1, Ordering::SeqCst) == 0 {
          let condition = SortKeyCondition::BeginsWith(<P2 as SortKeyOf<
            TestPartitions,
          >>::wrap_sort_key(
            PREFIX.to_string()
          ));
          let sched_inner = sched.clone();
          let run_count = run_count.clone();
          let log = log.clone();
          let rearmed = rearmed.clone();
          // (3) The thunk captures the ORIGINAL keys, NOT the P2 key that will
          // fire it — this is the replay semantics under test.
          let updated = updated.clone();
          let deleted = deleted.clone();
          sched.register_pending_redispatch(
            crate::Ident::new("test-owner"),
            P2::KEY,
            condition,
            Box::new(move || {
              spawn_watcher_run_with_keys(
                sched_inner.clone(),
                updated.clone(),
                deleted.clone(),
                run_count.clone(),
                log.clone(),
                rearmed.clone(),
              );
            }),
          );
        }

        None
      }
    },
    DEFAULT_LANE,
  );
}

/// HARD PROOF (GLD-0035) that a cross-source-P2-triggered re-dispatch replays
/// the ORIGINAL P1 matched keys, and never substitutes the P2 trigger key.
#[apply(smol_macros::test!)]
async fn replayed_args_are_original_p1_keys() {
  let scheduler = new_scheduler();
  let run_count = Arc::new(AtomicUsize::new(0));
  let rearmed = Arc::new(AtomicUsize::new(0));
  let log = Arc::new(Mutex::new(MatchedKeyLog::default()));

  // (a) First trigger: a real P1 commit. Capture the exact inserted key(s) the
  // commit produced — these are the matched keys `on_new_chunk` would feed to
  // the watcher.
  let p1_keys = commit_p1_keys(
    &scheduler,
    "p1-source",
    "module-a",
    module_record("module-a"),
  );
  assert_eq!(p1_keys.len(), 1, "exactly one P1 key inserted");
  let original_p1_key = p1_keys[0].clone();
  assert_eq!(original_p1_key.partition_key(), P1::KEY.ident());

  spawn_watcher_run_with_keys(
    scheduler.clone(),
    p1_keys.clone(),
    Vec::new(),
    run_count.clone(),
    log.clone(),
    rearmed.clone(),
  );

  assert!(
    wait_for(
      || run_count.load(Ordering::SeqCst) >= 1,
      Duration::from_secs(5)
    )
    .await,
    "watcher did not run after P1 trigger"
  );
  let _ = wait_for(|| false, Duration::from_millis(50)).await;
  assert_eq!(
    run_count.load(Ordering::SeqCst),
    1,
    "exactly one run after the P1 trigger, before any P2 write"
  );

  // (b) A DIFFERENT source writes a NEW P2 row under the prefix. P1 is never
  // touched. This routes through on_new_chunk -> fire_pending_redispatch and
  // re-runs the watcher. The P2 sort key is distinct from the P1 key.
  let p2_trigger_sort = format!("{PREFIX}first");
  commit_p2(
    &scheduler,
    "p2-other-source",
    &p2_trigger_sort,
    module_record("dep-first"),
  );
  let p2_trigger_key = crate::database::RecordKey::<TestPartitions>::new(
    P2::KEY,
    <P2 as SortKeyOf<TestPartitions>>::wrap_sort_key(p2_trigger_sort.clone()),
  );

  // (c) Wait for the re-run.
  assert!(
    wait_for(
      || run_count.load(Ordering::SeqCst) >= 2,
      Duration::from_secs(2),
    )
    .await,
    "watcher did not re-run after the cross-source P2 write"
  );
  let _ = wait_for(|| false, Duration::from_millis(100)).await;
  assert_eq!(
    run_count.load(Ordering::SeqCst),
    2,
    "exactly two runs: the P1 trigger and the single P2-triggered re-run"
  );

  // -- Assertions: the replay carried the ORIGINAL P1 keys ------------------
  let log = log.lock();
  assert_eq!(log.updated_per_run.len(), 2, "two recorded runs");

  // run[0]: matched-updated == the original P1 trigger key.
  assert_eq!(
    log.updated_per_run[0],
    vec![original_p1_key.clone()],
    "first run's matched_keys_updated is the original P1 key"
  );
  assert!(
    log.deleted_per_run[0].is_empty(),
    "first run had no deleted matched keys"
  );

  // run[1]: SAME args replayed — still the original P1 key, NOT the P2 key.
  assert_eq!(
    log.updated_per_run[1], log.updated_per_run[0],
    "the P2-triggered re-run replays the SAME matched keys as the first run"
  );
  assert_eq!(
    log.updated_per_run[1],
    vec![original_p1_key.clone()],
    "the re-run still carries the original P1 key"
  );

  // The P2 trigger key must NOT appear anywhere in the replayed matched keys.
  assert!(
    !log.updated_per_run[1].contains(&p2_trigger_key),
    "the P2 trigger key must not leak into the replayed matched keys"
  );
  assert!(
    !log.deleted_per_run[1].contains(&p2_trigger_key),
    "the P2 trigger key must not leak into the replayed deleted keys"
  );

  eprintln!(
    "run[0] matched_updated key_ident={} | run[1] matched_updated key_ident={} \
     | p2_trigger key_ident={}",
    log.updated_per_run[0][0].key_ident(),
    log.updated_per_run[1][0].key_ident(),
    p2_trigger_key.key_ident(),
  );
}