fix(data): recover queries after scheduler cancellation or rejection

docs/core/data.md

@@ -175,6 +175,16 @@ load only, while `refreshing` and `pending-write` always imply `stale: true` and
 previous value available. A mutation-driven invalidation commits `pending-write` first, then
 moves to `refreshing` when the confirming fetch starts.
 `staleReason` narrows settled stale states into `inconsistent`, `aborted`, or `error`.
+If the scheduler clears or rejects a queued query before its fetch starts, the
+associated refresh promise settles without fetching or replacing the current data.
+A later refresh can schedule normally. Development admission errors still throw
+synchronously; cancellation also settles a shared promise when it owns a queued
+invalidation replacement or reconciliation retry. Clearing queued work does not
+publish a new query state: if invalidation already aborted a running request, its
+last snapshot can still report `refreshing` until a later explicit refresh.
+Late results or errors from the aborted request cannot replace that snapshot,
+even if its fetch ignores the abort signal.
+
 Manual calls to `refresh()` coalesce while a request is pending. `invalidate()`
 is the distinct operation that replaces stale work; rapid invalidations before
 the replacement begins coalesce into the latest queued refresh. A `reconcile` callback may

src/data/query-cell.ts

@@ -8,7 +8,11 @@ import {
 import { getComponentLifetimeIdentity } from '../runtime/component/capabilities';
 import { logger } from '../common/logger';
 import { getActiveRenderContext } from '../common/render-context';
-import { adjustOwnershipDiagnostic, enqueueRuntimeTask } from '../runtime';
+import {
+  adjustOwnershipDiagnostic,
+  requestRuntimeWork,
+  ScheduledWork,
+} from '../runtime';
 import {
   claimHookIndex,
   getCurrentAppRenderRuntime,
@@ -43,14 +47,27 @@ type QueryCellOptions<T> = QueryOptions<T> & {
   readonly definitionIdentity?: object;
 };
 
+class QueryStartWork extends ScheduledWork {
+  constructor(
+    run: () => void,
+    private readonly settle: () => void
+  ) {
+    super(run);
+  }
+
+  protected override cancel(): void {
+    super.cancel();
+    this.settle();
+  }
+}
+
 export class QueryCell<T> {
   private readonly source = createReadableSource();
   private readonly key: string;
   private readonly cache: Map<string, QueryCell<unknown>>;
   private options: QueryCellOptions<T>;
   private controller: AbortController | null = null;
   private generation = 0;
-  private startQueued = false;
   private pendingRefresh: Promise<void> | null = null;
   private pendingRefreshKind:
     | 'initial'
@@ -158,7 +175,6 @@ export class QueryCell<T> {
     }
     this.controller?.abort();
     this.controller = null;
-    this.startQueued = false;
     this.reconcileAttemptCount = 0;
     this.ownerCount = 0;
     this.owners.clear();
@@ -230,8 +246,7 @@ export class QueryCell<T> {
       !this.destroyed &&
       this.state.data === null &&
       this.state.error === null &&
-      !this.pendingRefresh &&
-      !this.startQueued
+      !this.pendingRefresh
     );
   }
 
@@ -240,7 +255,6 @@ export class QueryCell<T> {
       this.destroyed ||
       this.state.data !== null ||
       this.pendingRefresh ||
-      this.startQueued ||
       this.options.skipInitialFetch
     ) {
       return;
@@ -258,7 +272,7 @@ export class QueryCell<T> {
       if (this.pendingRefreshKind === 'invalidation') {
         this.controller?.abort();
         this.queueStart(undefined, 'manual', true);
-        return this.pendingRefresh;
+        return this.pendingRefresh ?? Promise.resolve();
       } else {
         return this.pendingRefresh;
       }
@@ -311,27 +325,37 @@ export class QueryCell<T> {
         this.reconcileAttemptCount = 0;
         return ++this.reconcileSequence;
       })();
-    this.startQueued = true;
     this.pendingRefreshKind = kind;
     const token = ++this.pendingRefreshToken;
     if (!continuePending || !this.pendingRefresh) {
       this.pendingRefresh = new Promise<void>((resolve) => {
         this.pendingRefreshResolve = resolve;
       });
     }
-    enqueueRuntimeTask(() => {
-      if (token !== this.pendingRefreshToken) {
-        return;
-      }
-      this.startQueued = false;
-      if (this.destroyed) {
-        this.finishPendingRefresh(token);
-        return;
-      }
-      void this.start(sequence).finally(() => {
-        this.finishPendingRefresh(token);
-      });
-    });
+    requestRuntimeWork(
+      'component',
+      new QueryStartWork(
+        () => {
+          if (token !== this.pendingRefreshToken) {
+            return;
+          }
+          if (this.destroyed) {
+            this.finishPendingRefresh(token);
+            return;
+          }
+          void this.start(sequence).finally(() => {
+            this.finishPendingRefresh(token);
+          });
+        },
+        () => {
+          if (token !== this.pendingRefreshToken) return;
+          // The replacement may have already aborted a running fetch. Retire
+          // its authority even when that fetch ignores cancellation.
+          this.generation += 1;
+          this.finishPendingRefresh(token);
+        }
+      )
+    );
   }
 
   private finishPendingRefresh(token = this.pendingRefreshToken): void {

src/runtime/access.ts

@@ -10,6 +10,7 @@ import type {
 } from './renderer-capabilities';
 import type { Scheduler, SchedulerLane } from './scheduler';
 import type { ScheduledWork } from './scheduled-work';
+export { ScheduledWork } from './scheduled-work';
 export { SCHEDULER_LANES } from './scheduler';
 import {
   clearCurrentComponentScope,

src/runtime/scheduled-work.ts

@@ -22,13 +22,17 @@ export class ScheduledWork {
     try {
       scheduler.enqueueInLane(lane, this.task);
     } catch (error) {
-      this.pending = false;
+      this.cancel();
       throw error;
     }
   }
 
+  // Overrides may only reset internal ownership, never throw or call user code.
+  protected cancel(): void {
+    this.pending = false;
+  }
+
   static release(task: () => void): void {
-    const work = workByTask.get(task);
-    if (work) work.pending = false;
+    workByTask.get(task)?.cancel();
   }
 }

tests/unit/data/query-admission.test.ts

@@ -0,0 +1,210 @@
+import { afterEach, describe, expect, it, vi } from 'vite-plus/test';
+import { createDataRuntime, createQuery, invalidate } from '../../../src/data';
+import * as environment from '../../../src/common/env';
+import { globalScheduler as scheduler } from '../../../src/runtime/scheduler';
+
+afterEach(() => {
+  scheduler.setBulkCommitProbe(() => false);
+  scheduler.clearPendingSyncTasks();
+  vi.useRealTimers();
+  vi.restoreAllMocks();
+});
+
+describe('query work admission', () => {
+  it.each(['initial', 'manual', 'invalidation'] as const)(
+    'should settle cleared %s work and admits a later refresh',
+    async (kind) => {
+      const runtime = createDataRuntime();
+      const fetch = vi.fn(async () => ({ value: 2 }));
+      const query = createQuery({
+        key: 'admission',
+        runtime,
+        fetch,
+        skipInitialFetch: kind !== 'initial',
+      });
+      if (kind === 'invalidation') invalidate('admission', { runtime });
+      const pending = query.refresh();
+      let settled = false;
+      void pending.then(() => {
+        settled = true;
+      });
+      scheduler.clearPendingSyncTasks();
+      await Promise.resolve();
+      expect(settled).toBe(true);
+      expect(fetch).not.toHaveBeenCalled();
+      const next = query.refresh();
+      scheduler.flush();
+      await next;
+      expect(query.data).toEqual({ value: 2 });
+      expect(fetch).toHaveBeenCalledTimes(1);
+    }
+  );
+
+  it('should preserve the last snapshot when an aborted request replacement is cleared', async () => {
+    const runtime = createDataRuntime();
+    let signal!: AbortSignal;
+    const fetch = vi.fn(({ signal: nextSignal }: { signal: AbortSignal }) => {
+      signal = nextSignal;
+      return fetch.mock.calls.length === 1
+        ? new Promise<{ value: number }>(() => {})
+        : Promise.resolve({ value: 2 });
+    });
+    const query = createQuery({
+      key: 'replacement',
+      runtime,
+      fetch,
+      initialData: { value: 1 },
+    });
+    let settled = false;
+    void query.refresh().then(() => {
+      settled = true;
+    });
+    scheduler.flush();
+    expect(query.refreshing).toBe(true);
+    invalidate('replacement', { runtime });
+    expect(signal.aborted).toBe(true);
+    scheduler.clearPendingSyncTasks();
+    await Promise.resolve();
+    expect(settled).toBe(true);
+    expect(query.data).toEqual({ value: 1 });
+    expect(query.refreshing).toBe(true);
+    const next = query.refresh();
+    scheduler.flush();
+    await next;
+    expect(query.data).toEqual({ value: 2 });
+    expect(query.refreshing).toBe(false);
+  });
+
+  for (const interruption of [
+    'clear',
+    'reject',
+    'production-reject',
+  ] as const) {
+    it.each(['resolve', 'reject'] as const)(
+      `should ignore late %s after replacement ${interruption}`,
+      async (outcome) => {
+        const runtime = createDataRuntime();
+        let resolve!: (value: { value: number }) => void;
+        let reject!: (reason: Error) => void;
+        const query = createQuery({
+          key: 'late',
+          runtime,
+          initialData: { value: 1 },
+          fetch: () =>
+            new Promise<{ value: number }>((res, rej) => {
+              resolve = res;
+              reject = rej;
+            }),
+        });
+        const pending = query.refresh();
+        scheduler.flush();
+        if (interruption === 'clear') {
+          invalidate('late', { runtime });
+          scheduler.clearPendingSyncTasks();
+        } else {
+          if (interruption === 'production-reject') {
+            vi.spyOn(environment, 'isDevelopmentEnvironment').mockReturnValue(
+              false
+            );
+          }
+          scheduler.setBulkCommitProbe(() => true);
+          if (interruption === 'reject') {
+            expect(() => invalidate('late', { runtime })).toThrow(
+              'during bulk commit'
+            );
+          } else invalidate('late', { runtime });
+          scheduler.setBulkCommitProbe(() => false);
+        }
+        await pending;
+        if (outcome === 'resolve') resolve({ value: 99 });
+        else reject(new Error('obsolete failure'));
+        await Promise.resolve();
+        await Promise.resolve();
+        expect(query.data).toEqual({ value: 1 });
+        expect(query.error).toBeNull();
+        expect(query.refreshing).toBe(true);
+      }
+    );
+  }
+
+  it('should settle silently rejected production work and recovers', async () => {
+    vi.spyOn(environment, 'isDevelopmentEnvironment').mockReturnValue(false);
+    const fetch = vi.fn(async () => ({ value: 2 }));
+    const query = createQuery({
+      key: 'production',
+      runtime: createDataRuntime(),
+      fetch,
+      skipInitialFetch: true,
+    });
+    scheduler.setBulkCommitProbe(() => true);
+    await query.refresh();
+    scheduler.setBulkCommitProbe(() => false);
+    const next = query.refresh();
+    scheduler.flush();
+    await next;
+    expect(fetch).toHaveBeenCalledTimes(1);
+  });
+
+  it('should settle the shared promise when manual promotion is rejected', async () => {
+    const runtime = createDataRuntime();
+    const query = createQuery({
+      key: 'promoted',
+      runtime,
+      fetch: async () => ({ value: 2 }),
+      skipInitialFetch: true,
+    });
+    const original = query.refresh();
+    invalidate('promoted', { runtime });
+    vi.spyOn(environment, 'isDevelopmentEnvironment').mockReturnValue(false);
+    scheduler.setBulkCommitProbe(() => true);
+    const promoted = query.refresh();
+    expect(promoted).toBeInstanceOf(Promise);
+    await promoted;
+    await original;
+  });
+
+  it('should recover after development admission rejection', async () => {
+    const fetch = vi.fn(async () => ({ value: 2 }));
+    const query = createQuery({
+      key: 'rejected',
+      runtime: createDataRuntime(),
+      fetch,
+      skipInitialFetch: true,
+    });
+    scheduler.setBulkCommitProbe(() => true);
+    expect(() => query.refresh()).toThrow('during bulk commit');
+    scheduler.setBulkCommitProbe(() => false);
+    const next = query.refresh();
+    scheduler.flush();
+    expect(fetch).toHaveBeenCalledTimes(1);
+    await next;
+    expect(query.data).toEqual({ value: 2 });
+  });
+
+  it('should settle a refresh when its reconciliation retry is cleared', async () => {
+    vi.useFakeTimers();
+    const fetch = vi.fn(async () => ({ value: 2 }));
+    const query = createQuery({
+      key: 'reconcile',
+      runtime: createDataRuntime(),
+      fetch,
+      skipInitialFetch: true,
+      isConsistent: () => false,
+      reconcile: () => true,
+    });
+    let settled = false;
+    void query.refresh().then(() => {
+      settled = true;
+    });
+    scheduler.flush();
+    const enqueue = scheduler.enqueueInLane.bind(scheduler);
+    vi.spyOn(scheduler, 'enqueueInLane').mockImplementation((lane, task) => {
+      enqueue(lane, task);
+      scheduler.clearPendingSyncTasks();
+    });
+    await vi.advanceTimersByTimeAsync(25);
+    await Promise.resolve();
+    expect(settled).toBe(true);
+    expect(query.stale).toBe(true);
+  });
+});