perf(zero-cache): add head-indexed queue helpers

packages/shared/src/head-indexed-queue.test.ts

@@ -0,0 +1,55 @@
+import {describe, expect, test} from 'vitest';
+import {HeadIndexedQueue} from './head-indexed-queue.ts';
+
+describe('HeadIndexedQueue', () => {
+  test('keeps FIFO order across many shifts without exposing compacted slots', () => {
+    const queue = new HeadIndexedQueue<number>();
+    for (let i = 0; i < 2050; i++) {
+      queue.push(i);
+    }
+
+    for (let i = 0; i < 1500; i++) {
+      expect(queue.shift()).toBe(i);
+    }
+    queue.push(2050);
+
+    expect(queue.size).toBe(551);
+    expect(queue.peek()).toBe(1500);
+    expect(queue.last()).toBe(2050);
+    expect(queue.toArray()).toEqual(
+      Array.from({length: 551}, (_, i) => i + 1500),
+    );
+  });
+
+  test('delete and replace operations skip holes left in the active window', () => {
+    const queue = new HeadIndexedQueue<string>();
+    queue.push('a');
+    queue.push('b');
+    queue.push('c');
+    queue.push('d');
+
+    expect(queue.deleteFirst('a')).toBe(true);
+    expect(queue.deleteMatching(value => value === 'c')).toBe(1);
+    queue.replaceLast('D');
+
+    expect(queue.size).toBe(2);
+    expect(queue.peek()).toBe('b');
+    expect(queue.last()).toBe('D');
+    expect(queue.toArray()).toEqual(['b', 'D']);
+    expect(queue.shift()).toBe('b');
+    expect(queue.shift()).toBe('D');
+    expect(queue.shift()).toBeUndefined();
+  });
+
+  test('stores undefined values without confusing them with deleted slots', () => {
+    const queue = new HeadIndexedQueue<string | undefined>();
+    queue.push(undefined);
+    queue.push('next');
+
+    expect(queue.size).toBe(2);
+    expect(queue.shift()).toBeUndefined();
+    expect(queue.size).toBe(1);
+    expect(queue.shift()).toBe('next');
+    expect(queue.size).toBe(0);
+  });
+});

packages/shared/src/head-indexed-queue.ts

@@ -0,0 +1,143 @@
+// High-volume RM -> VS streams should not make every FIFO dequeue move the
+// rest of the array. Keep the head-index/compaction bookkeeping in one small
+// type so hot queues can stay O(1) without spreading that machinery through
+// the caller.
+export class HeadIndexedQueue<T> {
+  // `undefined` is a valid queue value for some callers, so deleted slots use a
+  // private sentinel instead of overloading undefined.
+  readonly #entries: (T | EmptySlot)[] = [];
+  // #head is the first slot worth scanning; #count is the number of live values
+  // after holes from deletes/shifts. Keeping both lets shift() stay O(1) while
+  // deleteMatching() can remove arbitrary pending entries.
+  #head = 0;
+  #count = 0;
+
+  get size(): number {
+    return this.#count;
+  }
+
+  push(value: T): void {
+    this.#entries.push(value);
+    this.#count++;
+  }
+
+  peek(): T | undefined {
+    if (this.#count === 0) {
+      return undefined;
+    }
+    for (let i = this.#head; i < this.#entries.length; i++) {
+      const value = this.#entries[i];
+      if (value !== EMPTY_SLOT) {
+        return value;
+      }
+    }
+    return undefined;
+  }
+
+  shift(): T | undefined {
+    if (this.#count === 0) {
+      this.clear();
+      return undefined;
+    }
+
+    while (this.#head < this.#entries.length) {
+      const value = this.#entries[this.#head];
+      // Clear the reference before returning so long-lived streams do not keep
+      // consumed payloads alive until the next compaction.
+      this.#entries[this.#head] = EMPTY_SLOT;
+      this.#head++;
+      if (value !== EMPTY_SLOT) {
+        this.#count--;
+        this.#maybeCompact();
+        return value;
+      }
+    }
+
+    this.clear();
+    return undefined;
+  }
+
+  last(): T | undefined {
+    if (this.#count === 0) {
+      return undefined;
+    }
+    for (let i = this.#entries.length - 1; i >= this.#head; i--) {
+      const value = this.#entries[i];
+      if (value !== EMPTY_SLOT) {
+        return value;
+      }
+    }
+    return undefined;
+  }
+
+  replaceLast(value: T): void {
+    for (let i = this.#entries.length - 1; i >= this.#head; i--) {
+      if (this.#entries[i] !== EMPTY_SLOT) {
+        this.#entries[i] = value;
+        return;
+      }
+    }
+    this.push(value);
+  }
+
+  deleteFirst(value: T): boolean {
+    for (let i = this.#head; i < this.#entries.length; i++) {
+      if (this.#entries[i] === value) {
+        this.#entries[i] = EMPTY_SLOT;
+        this.#count--;
+        this.#maybeCompact();
+        return true;
+      }
+    }
+    return false;
+  }
+
+  deleteMatching(predicate: (value: T) => boolean): number {
+    let deleted = 0;
+    // Scan backward so several deletions can create holes without changing the
+    // indexes of entries that have not been inspected yet.
+    for (let i = this.#entries.length - 1; i >= this.#head; i--) {
+      const value = this.#entries[i];
+      if (value !== EMPTY_SLOT && predicate(value)) {
+        this.#entries[i] = EMPTY_SLOT;
+        this.#count--;
+        deleted++;
+      }
+    }
+    this.#maybeCompact();
+    return deleted;
+  }
+
+  toArray(): T[] {
+    const values: T[] = [];
+    for (let i = this.#head; i < this.#entries.length; i++) {
+      const value = this.#entries[i];
+      if (value !== EMPTY_SLOT) {
+        values.push(value);
+      }
+    }
+    return values;
+  }
+
+  clear(): void {
+    this.#entries.length = 0;
+    this.#head = 0;
+    this.#count = 0;
+  }
+
+  #maybeCompact(): void {
+    if (this.#count === 0) {
+      // Empty queues should release the whole backing array immediately; this is
+      // the common state after a stream drains.
+      this.clear();
+    } else if (this.#head > 1024 && this.#head * 2 > this.#entries.length) {
+      // Compaction copies the live suffix, so wait until the dead prefix is both
+      // large and at least half the array.
+      this.#entries.splice(0, this.#head);
+      this.#head = 0;
+    }
+  }
+}
+
+const EMPTY_SLOT = Symbol('empty-slot');
+type EmptySlot = typeof EMPTY_SLOT;

packages/zero-cache/src/types/subscription.test.ts

@@ -628,4 +628,79 @@ describe('types/subscription', () => {
     });
     expect(subWithCoalesceAndPipeline.pipeline).not.toBeUndefined();
   });
+
+  test('pipeline cancel cleanup ignores already dequeued head entries', async () => {
+    const consumed: number[] = [];
+    const cleanup = vi.fn();
+    const results: Promise<Result>[] = [];
+
+    const subscription = Subscription.create<number>({
+      cleanup,
+      consumed: m => consumed.push(m),
+    });
+    for (let i = 0; i < 1500; i++) {
+      results.push(subscription.push(i).result);
+    }
+
+    assert(
+      subscription.pipeline,
+      'Expected subscription pipeline to be defined',
+    );
+    const iterator = subscription.pipeline[Symbol.asyncIterator]();
+
+    for (let i = 0; i < 1200; i++) {
+      const next = await iterator.next();
+      assert(!next.done, 'Expected next subscription entry');
+      expect(next.value.value).toBe(i);
+      next.value.consumed();
+      expect(await results[i]).toBe('consumed');
+    }
+
+    const current = await iterator.next();
+    assert(!current.done, 'Expected current subscription entry');
+    expect(current.value.value).toBe(1200);
+    expect(subscription.queued).toBe(299);
+
+    subscription.cancel();
+    expect(await iterator.next()).toEqual({value: undefined, done: true});
+
+    for (let i = 1200; i < 1500; i++) {
+      expect(await results[i]).toBe('unconsumed');
+    }
+    expect(consumed).toEqual(Array.from({length: 1200}, (_, i) => i));
+    expect(cleanup).toBeCalledTimes(1);
+    expect(cleanup.mock.calls[0][0]).toEqual(
+      Array.from({length: 300}, (_, i) => i + 1200),
+    );
+  });
+
+  test('end drains queued messages after head advances', async () => {
+    const consumed: number[] = [];
+    const cleanup = vi.fn();
+    const results: Promise<Result>[] = [];
+
+    const subscription = Subscription.create<number>({
+      cleanup,
+      consumed: m => consumed.push(m),
+    });
+    for (let i = 0; i < 1500; i++) {
+      results.push(subscription.push(i).result);
+    }
+
+    const received: number[] = [];
+    for await (const m of subscription) {
+      received.push(m);
+      if (m === 1199) {
+        subscription.end();
+      }
+    }
+
+    expect(received).toEqual(Array.from({length: 1500}, (_, i) => i));
+    expect(consumed).toEqual(received);
+    expect(cleanup).toBeCalledTimes(1);
+    expect(cleanup.mock.calls[0][0]).toEqual([]);
+    for (const result of results) {
+      expect(await result).toBe('consumed');
+    }
+  });
 });

packages/zero-cache/src/types/subscription.ts

@@ -1,5 +1,6 @@
 import {resolver, type Resolver} from '@rocicorp/resolver';
 import {assert} from '../../../shared/src/asserts.ts';
+import {HeadIndexedQueue} from '../../../shared/src/head-indexed-queue.ts';
 import type {Sink, Source} from './streams.ts';
 
 /**
@@ -74,10 +75,8 @@ export class Subscription<T, M = T> implements Source<T>, Sink<M> {
     return new Subscription(options, publish);
   }
 
-  // Consumers waiting to consume messages (i.e. an async iteration awaiting the next message).
-  readonly #consumers: Resolver<Entry<M> | null>[] = [];
-  // Messages waiting to be dequeued.
-  readonly #messages: (Entry<M> | 'terminus')[] = [];
+  readonly #consumers = new HeadIndexedQueue<Resolver<Entry<M> | null>>();
+  readonly #messages = new HeadIndexedQueue<Entry<M> | Terminus>();
   // Messages dequeued but not yet consumed.
   readonly #consuming = new Set<Entry<M>>();
   readonly #pipelineEnabled: boolean;
@@ -156,16 +155,15 @@ export class Subscription<T, M = T> implements Source<T>, Sink<M> {
       consumer.resolve(entry);
     } else if (
       this.#coalesce &&
-      this.#messages.length &&
-      this.#messages.at(-1) !== 'terminus'
+      this.#messages.size > 0 &&
+      this.#messages.last() !== TERMINUS
     ) {
-      // oxlint-disable-next-line typescript/no-non-null-assertion
-      const prev = this.#messages.at(-1)!;
-      assert(prev !== 'terminus', 'prev should not be terminus after check');
-      this.#messages[this.#messages.length - 1] = {
+      const prev = this.#messages.last();
+      assert(prev !== undefined && prev !== TERMINUS, 'expected an entry');
+      this.#messages.replaceLast({
         value: this.#coalesce(entry, prev),
         resolve,
-      };
+      });
     } else {
       this.#messages.push(entry);
     }
@@ -179,7 +177,7 @@ export class Subscription<T, M = T> implements Source<T>, Sink<M> {
 
   /** The number of messages waiting to be dequeued. */
   get queued(): number {
-    return this.#messages.length;
+    return this.#messages.size;
   }
 
   /** The number of messages dequeued but not yet "consumed" */
@@ -202,10 +200,10 @@ export class Subscription<T, M = T> implements Source<T>, Sink<M> {
   end() {
     if (this.#sentinel) {
       // already terminated
-    } else if (this.#messages.length === 0) {
+    } else if (this.#messages.size === 0) {
       this.cancel();
     } else {
-      this.#messages.push('terminus');
+      this.#messages.push(TERMINUS);
     }
   }
 
@@ -230,20 +228,23 @@ export class Subscription<T, M = T> implements Source<T>, Sink<M> {
   #terminate(sentinel: 'canceled' | Error) {
     if (!this.#sentinel) {
       this.#sentinel = sentinel;
+      const pendingMessages = this.#messages.toArray().filter(isEntry);
       this.#cleanup(
-        [...this.#consuming, ...this.#messages.filter(m => m !== 'terminus')],
+        [...this.#consuming, ...pendingMessages],
         sentinel instanceof Error ? sentinel : undefined,
       );
-      this.#messages.splice(0);
+      this.#messages.clear();
 
       for (
         let consumer = this.#consumers.shift();
-        consumer;
+        consumer !== undefined;
         consumer = this.#consumers.shift()
       ) {
-        sentinel === 'canceled'
-          ? consumer.resolve(null)
-          : consumer.reject(sentinel);
+        if (sentinel === 'canceled') {
+          consumer.resolve(null);
+        } else {
+          consumer.reject(sentinel);
+        }
       }
     }
   }
@@ -258,7 +259,7 @@ export class Subscription<T, M = T> implements Source<T>, Sink<M> {
     return {
       next: async () => {
         const entry = this.#messages.shift();
-        if (entry === 'terminus') {
+        if (entry === TERMINUS) {
           this.cancel();
           return {value: undefined, done: true};
         }
@@ -393,3 +394,10 @@ type Entry<M> = {
   readonly value: M;
   readonly resolve: (r: Result) => void;
 };
+
+const TERMINUS = Symbol('terminus');
+type Terminus = typeof TERMINUS;
+
+function isEntry<M>(entry: Entry<M> | Terminus): entry is Entry<M> {
+  return entry !== TERMINUS;
+}