upload current sources

worker/test/arenaDeltaPoll.test.ts

@@ -0,0 +1,134 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { HealthRegistry } from '../src/health/registry.js';
+import { IngestBufferFullError, makeArenaDeltaPoll } from '../src/poller/arenaDeltaPoll.js';
+import { DeltaPoller } from '../src/poller/deltaPoller.js';
+import { InMemoryEventQueue } from '../src/queue/webhookQueue.js';
+
+describe('makeArenaDeltaPoll', () => {
+  it('fetches changes since the cursor and enqueues each as a poll-sourced event', async () => {
+    const queue = new InMemoryEventQueue({ maxSize: 10 });
+    const fetchChangesSince = vi.fn().mockResolvedValue([{ id: 'c1' }, { id: 'c2' }]);
+    const warn = vi.fn();
+    const poll = makeArenaDeltaPoll({ client: { fetchChangesSince }, queue, log: { warn } });
+
+    const since = new Date('2026-09-14T08:00:00Z');
+    await poll(since);
+
+    expect(fetchChangesSince).toHaveBeenCalledWith(since);
+    expect(queue.peekAll().map((e) => [e.source, e.payload])).toEqual([
+      ['poll', { id: 'c1' }],
+      ['poll', { id: 'c2' }],
+    ]);
+    expect(warn).not.toHaveBeenCalled();
+  });
+
+  it('fails the tick WITHOUT calling Arena when the buffer is already full (B1)', async () => {
+    const queue = new InMemoryEventQueue({ maxSize: 1 });
+    queue.enqueue({ source: 'webhook', payload: { n: 1 } });
+    const fetchChangesSince = vi.fn().mockResolvedValue([{ id: 'c1' }]);
+    const warn = vi.fn();
+    const poll = makeArenaDeltaPoll({ client: { fetchChangesSince }, queue, log: { warn } });
+
+    await expect(poll(new Date(0))).rejects.toBeInstanceOf(IngestBufferFullError);
+
+    expect(fetchChangesSince).not.toHaveBeenCalled();
+    expect(queue.size).toBe(1);
+    expect(warn).toHaveBeenCalledTimes(1);
+    expect(warn.mock.calls[0]?.[1]).toMatch(/queue full/);
+  });
+
+  it('fails the tick after a partial enqueue when the buffer fills mid-poll (B1)', async () => {
+    const queue = new InMemoryEventQueue({ maxSize: 1 });
+    const fetchChangesSince = vi.fn().mockResolvedValue([{ id: 'c1' }, { id: 'c2' }, { id: 'c3' }]);
+    const warn = vi.fn();
+    const poll = makeArenaDeltaPoll({ client: { fetchChangesSince }, queue, log: { warn } });
+
+    const failure = await poll(new Date(0)).catch((err: unknown) => err);
+    expect(failure).toBeInstanceOf(IngestBufferFullError);
+    expect(failure).toMatchObject({ fetched: 3, dropped: 2 });
+
+    // What fit stays queued; the rest is re-covered by the next tick (same window).
+    expect(queue.size).toBe(1);
+    expect(queue.peekAll()[0]?.payload).toEqual({ id: 'c1' });
+    expect(warn).toHaveBeenCalledTimes(1);
+    expect(warn.mock.calls[0]?.[0]).toMatchObject({ dropped: 2, fetched: 3 });
+  });
+
+  it('propagates a fetch failure so the poller keeps its cursor', async () => {
+    const queue = new InMemoryEventQueue({ maxSize: 1 });
+    const fetchChangesSince = vi.fn().mockRejectedValue(new Error('arena unreachable'));
+    const poll = makeArenaDeltaPoll({ client: { fetchChangesSince }, queue, log: { warn: vi.fn() } });
+
+    await expect(poll(new Date(0))).rejects.toThrow('arena unreachable');
+    expect(queue.size).toBe(0);
+  });
+});
+
+describe('makeArenaDeltaPoll driven by DeltaPoller (B1 end to end)', () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it('a partial drop holds the cursor, flips health to degraded, and the next tick re-covers the window', async () => {
+    const queue = new InMemoryEventQueue({ maxSize: 1 });
+    const fetchChangesSince = vi.fn().mockResolvedValue([{ id: 'c1' }, { id: 'c2' }]);
+    const health = new HealthRegistry();
+    const start = new Date('2026-09-14T08:00:00Z');
+    let nowMs = start.getTime();
+    const poller = new DeltaPoller({
+      intervalMs: 1000,
+      poll: makeArenaDeltaPoll({ client: { fetchChangesSince }, queue, log: { warn: vi.fn() } }),
+      health,
+      now: () => new Date(nowMs),
+    });
+    const initialSince = poller.nextSince;
+
+    poller.start();
+    nowMs += 1000;
+    await vi.advanceTimersByTimeAsync(1000);
+
+    expect(fetchChangesSince).toHaveBeenCalledTimes(1);
+    expect(fetchChangesSince).toHaveBeenLastCalledWith(initialSince);
+    expect(poller.nextSince).toEqual(initialSince); // cursor did NOT advance
+    expect(health.status()).toBe('degraded');
+    expect(health.pollerSnapshot()?.lastError).toBe('Ingest buffer full at poll time.');
+
+    // Buffer drains; the next tick re-covers the SAME window and succeeds.
+    queue.dequeue();
+    fetchChangesSince.mockResolvedValue([{ id: 'c2' }]);
+    nowMs += 1000;
+    await vi.advanceTimersByTimeAsync(1000);
+
+    expect(fetchChangesSince).toHaveBeenCalledTimes(2);
+    expect(fetchChangesSince).toHaveBeenLastCalledWith(initialSince);
+    expect(poller.nextSince.getTime()).toBeGreaterThan(initialSince.getTime());
+    expect(health.status()).toBe('ok');
+
+    await poller.stop();
+  });
+
+  it('a full buffer before the fetch never calls Arena and leaves the cursor untouched', async () => {
+    const queue = new InMemoryEventQueue({ maxSize: 1 });
+    queue.enqueue({ source: 'webhook', payload: {} });
+    const fetchChangesSince = vi.fn().mockResolvedValue([]);
+    const health = new HealthRegistry();
+    const poller = new DeltaPoller({
+      intervalMs: 1000,
+      poll: makeArenaDeltaPoll({ client: { fetchChangesSince }, queue, log: { warn: vi.fn() } }),
+      health,
+    });
+    const initialSince = poller.nextSince;
+
+    poller.start();
+    await vi.advanceTimersByTimeAsync(2500);
+
+    expect(fetchChangesSince).not.toHaveBeenCalled();
+    expect(poller.nextSince).toEqual(initialSince);
+    expect(health.status()).toBe('degraded');
+    await poller.stop();
+  });
+});