feat: /polls plane — stateful polling jobs (WebFetchPollJob), scheduler, event forwarder - #6
Conversation
…re seam, config quorum Spec: cor:web:030:00/:02. The plane boots only when DATABASE_URL + TOKEN_ENCRYPTION_KEY + CORE_EVENTS_URL are all present; ops-only otherwise. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…tick, event forwarder Creation fail-fast (interval floor, mandatory TTL, org cap, guard-check the URL now, ops-plane header/auth policy), tick with expiry-first ordering, per-tick full guard, first-observation baseline init, fire-policy gating (once/every_change/cooldown, edge-triggered), advance-only-after-delivery, exponential backoff with terminal poll.failed at the limit. Forwarder throws on non-2xx (never log-and-swallow). Specs: cor:web:030:00..03. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Covers first-observation init, edge-triggered fire policies, advance-only- after-delivery with retry, backoff curve + poll.failed limit, expiry-first with accepted-transition, lease claiming, creation caps + guard fail-fast, credential never on a read surface, cancel-disposes-credential. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…ot wiring Prisma 6 (classic client, the stateful-tool archetype); postinstall generates the client so CI's npm ci keeps working. The plane boots only on the full env quorum — ops-only otherwise, unchanged. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a stateful polling plane to the microservice, adding Prisma-based database storage, a scheduler to run due poll jobs with exponential backoff, credential encryption at rest, and a set of internal /polls endpoints for job management. The review feedback highlights several critical improvements: catching errors in the scheduler's async interval loop to prevent unhandled promise rejections, consuming the response body in the event forwarder to avoid socket leaks, checking the scheduler lease before forcing a manual job run, and implementing pagination or limits on the organization job listing query to prevent performance bottlenecks.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| export function startScheduler(deps: SchedulerDeps, everyMs = 5_000): () => void { | ||
| let running = false; | ||
| const timer = setInterval(async () => { | ||
| if (running) return; // never overlap loop iterations | ||
| running = true; | ||
| try { | ||
| await runDueOnce(deps); | ||
| } finally { | ||
| running = false; | ||
| } | ||
| }, everyMs); | ||
| timer.unref?.(); | ||
| return () => clearInterval(timer); | ||
| } |
There was a problem hiding this comment.
The setInterval callback is an async function, but any error thrown by runDueOnce is not caught (there is only a try-finally block, no catch). In Node.js, this will result in an unhandled promise rejection, which can crash the process. Add a catch block to safely log the error.
export function startScheduler(deps: SchedulerDeps, everyMs = 5_000): () => void {
let running = false;
const timer = setInterval(async () => {
if (running) return; // never overlap loop iterations
running = true;
try {
await runDueOnce(deps);
} catch (err) {
logger.error('scheduler loop failed', {
err: String((err as Error)?.message ?? err).slice(0, 200),
});
} finally {
running = false;
}
}, everyMs);
timer.unref?.();
return () => clearInterval(timer);
}| export function createEventForwarder(options: ForwarderOptions): EventForwarder { | ||
| const { coreEventsUrl, coreEventsToken, timeoutMs = 10_000, fetchImpl = fetch } = options; | ||
| return async (event) => { | ||
| let response: Response; | ||
| try { | ||
| response = await fetchImpl(coreEventsUrl, { | ||
| method: 'POST', | ||
| headers: { | ||
| 'content-type': 'application/json', | ||
| ...(coreEventsToken ? { authorization: `Bearer ${coreEventsToken}` } : {}), | ||
| }, | ||
| body: JSON.stringify(event), | ||
| signal: AbortSignal.timeout(timeoutMs), | ||
| }); | ||
| } catch (err) { | ||
| throw new EventDeliveryError(String((err as Error)?.message ?? err).slice(0, 200)); | ||
| } | ||
| if (response.status < 200 || response.status >= 300) { | ||
| throw new EventDeliveryError(`core responded ${response.status}`); | ||
| } | ||
| }; | ||
| } |
There was a problem hiding this comment.
In Node.js 18+ (which uses undici under the hood for global fetch), not consuming the response body can prevent the socket from being released back to the keep-alive pool, potentially leading to socket leaks or connection exhaustion. It is recommended to consume the response body (e.g., via await response.text()) even for error responses.
export function createEventForwarder(options: ForwarderOptions): EventForwarder {
const { coreEventsUrl, coreEventsToken, timeoutMs = 10_000, fetchImpl = fetch } = options;
return async (event) => {
let response: Response;
try {
response = await fetchImpl(coreEventsUrl, {
method: 'POST',
headers: {
'content-type': 'application/json',
...(coreEventsToken ? { authorization: `Bearer ${coreEventsToken}` } : {}),
},
body: JSON.stringify(event),
signal: AbortSignal.timeout(timeoutMs),
});
} catch (err) {
throw new EventDeliveryError(String((err as Error)?.message ?? err).slice(0, 200));
}
await response.text().catch(() => {});
if (response.status < 200 || response.status >= 300) {
throw new EventDeliveryError(`core responded ${response.status}`);
}
};
}| router.post('/:id/run', async (req, res) => { | ||
| try { | ||
| const job = await service.store.get(req.params.id); | ||
| if (!job) { | ||
| res.status(404).json({ error: 'not_found', message: 'No such poll job.' }); | ||
| return; | ||
| } | ||
| if (job.status !== 'active') { | ||
| res.status(409).json({ error: 'poll_not_active', message: `Job is ${job.status}.` }); | ||
| return; | ||
| } | ||
| await tick(scheduler, job); | ||
| const after = await service.store.get(req.params.id); | ||
| res.json({ ok: true, job: after ? toView(after) : null }); | ||
| } catch (err) { | ||
| respondWithError(res, err, 'run'); | ||
| } | ||
| }); |
There was a problem hiding this comment.
The forced run endpoint POST /polls/:id/run executes tick directly on the job without checking if it is currently leased by the scheduler (leaseUntil). If the scheduler is concurrently processing the same job, this can lead to race conditions, double-fetching, or database write conflicts. Consider checking if the job is currently leased before running.
router.post('/:id/run', async (req, res) => {
try {
const job = await service.store.get(req.params.id);
if (!job) {
res.status(404).json({ error: 'not_found', message: 'No such poll job.' });
return;
}
if (job.status !== 'active') {
res.status(409).json({ error: 'poll_not_active', message: `Job is ${job.status}.` });
return;
}
const now = service.now();
if (job.leaseUntil && job.leaseUntil > now) {
res.status(409).json({ error: 'poll_leased', message: 'Job is currently leased by the scheduler.' });
return;
}
await tick(scheduler, job);
const after = await service.store.get(req.params.id);
res.json({ ok: true, job: after ? toView(after) : null });
} catch (err) {
respondWithError(res, err, 'run');
}
});| async listByOrg(orgId: string): Promise<PollJob[]> { | ||
| const rows = await this.prisma.webFetchPollJob.findMany({ where: { orgId }, orderBy: { createdAt: 'desc' } }); | ||
| return rows.map(toDomain); | ||
| } |
There was a problem hiding this comment.
The listByOrg query fetches all historical poll jobs for an organization without any limit or pagination. Over time, as completed, expired, failed, or cancelled jobs accumulate, this query can become a performance bottleneck and consume excessive memory. Consider adding pagination or a default limit (e.g., top 100 most recent).
async listByOrg(orgId: string): Promise<PollJob[]> {
const rows = await this.prisma.webFetchPollJob.findMany({
where: { orgId },
orderBy: { createdAt: 'desc' },
take: 100,
});
return rows.map(toDomain);
}![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 06ef5a2409
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| "test:watch": "vitest", | ||
| "test:coverage": "vitest run --coverage" | ||
| "test:coverage": "vitest run --coverage", | ||
| "postinstall": "prisma generate", |
There was a problem hiding this comment.
I checked the Dockerfile build order: it runs npm ci --include=dev immediately after copying only package.json and package-lock.json, while this new postinstall invokes prisma generate. Because prisma/schema.prisma is not present at that layer (and the Dockerfile never copies prisma/), image builds will fail before TypeScript compilation instead of producing a deployable service.
Useful? React with 👍 / 👎.
| if (!bestEffort) { | ||
| await recordFailure(deps, job, err); | ||
| return; |
There was a problem hiding this comment.
When an already-expired job cannot deliver poll.expired repeatedly, this calls the generic recordFailure, which applies the failure limit and eventually transitions the job to failed via a best-effort poll.failed. That contradicts the expiry-first contract in this same scheduler: core can permanently miss the poll.expired terminal event during a core outage, and the job stops retrying it after POLL_FAILURE_LIMIT.
Useful? React with 👍 / 👎.
| const input = createPollSchema.parse(raw); | ||
| validateConditions(input.conditions); |
There was a problem hiding this comment.
When a caller creates a poll with an explicit kind that conflicts with the conditions, such as contentKind: "json" plus a selector_exists condition, this only runs validateConditions and stores an active job. The scheduler later forces kind=json, so evaluateConditions rejects on every tick and the poll backs off until failed instead of returning a creation-time 400 like the evaluate-url path does.
Useful? React with 👍 / 👎.
| const active = await deps.store.countActiveByOrg(input.orgId); | ||
| if (active >= limits.maxActivePerOrg) { | ||
| throw new ValidationError('orgId', `the organization already has ${active} active polls (cap ${limits.maxActivePerOrg})`); |
There was a problem hiding this comment.
With concurrent POST /polls requests for the same org when the active count is below the cap, both requests can read the same countActiveByOrg value and then both insert jobs because the check is separate from store.create and there is no transaction, lock, or database constraint around the pair. In production this lets POLL_MAX_ACTIVE_PER_ORG be exceeded even though sequential tests pass.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Pull request overview
Adds the new stateful /polls polling plane to hadrontool-webfetch, including persistent poll jobs, a scheduler loop, and outbound event forwarding to core, while keeping /ops stateless and unchanged unless the polling env quorum is present.
Changes:
- Introduces
/pollsCRUD +POST /polls/:id/run, backed by aPollStoreseam (in-memory for tests/dev, Prisma/Postgres for prod). - Adds the polling scheduler tick loop with lease-based claiming, condition evaluation, fire policies, backoff, terminal events, and credential-at-rest encryption.
- Wires polling-plane boot behind env quorum and adds Prisma tooling/dependencies plus new test coverage.
Reviewed changes
Copilot reviewed 15 out of 16 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| src/routes/polls.ts | Express router for /polls endpoints (create/list/get/cancel/run) with error handling. |
| src/routes/polls.test.ts | Supertest coverage for /polls validation, credential non-leak, cancellation disposal, forced run, and unconfigured absence. |
| src/polls/store.ts | PollStore interface plus in-memory implementation used by tests/dev. |
| src/polls/service.ts | Poll lifecycle service: creation validation, fail-fast guard, encryption-at-create, cancellation disposal, and view mapping. |
| src/polls/scheduler.ts | Scheduler loop + tick() implementation: guarded fetch, evaluation, fire policies, delivery gating, backoff, and terminal transitions. |
| src/polls/scheduler.test.ts | Unit tests for tick invariants: baseline init, policies, advance-only-after-delivery, backoff/limit, expiry-first, and claiming. |
| src/polls/prismaStore.ts | Prisma/Postgres PollStore implementation including atomic claimDue SQL with SKIP LOCKED. |
| src/polls/forwarder.ts | Tool→core event forwarder that throws on any non-2xx / delivery failure. |
| src/polls/crypto.ts | AES-256-GCM credential encryption/decryption with key-id fingerprinting. |
| src/ops/index.ts | Exports header allowlist/auth schema/header normalization for reuse by polling service. |
| src/index.ts | Boots polling plane conditionally (env quorum), instantiates Prisma store/cipher/forwarder, starts scheduler. |
| src/config.ts | Adds polling-plane env vars + derived polling config/quorum gating. |
| src/app.ts | Conditionally mounts /polls router when polling deps are provided. |
| prisma/schema.prisma | Adds WebFetchPollJob Prisma model for persisted poll jobs. |
| package.json | Adds Prisma dependencies and scripts (postinstall, db:push). |
| package-lock.json | Lockfile updates for Prisma + transitive dependencies. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| router.post('/:id/run', async (req, res) => { | ||
| try { | ||
| const job = await service.store.get(req.params.id); | ||
| if (!job) { | ||
| res.status(404).json({ error: 'not_found', message: 'No such poll job.' }); | ||
| return; | ||
| } | ||
| if (job.status !== 'active') { | ||
| res.status(409).json({ error: 'poll_not_active', message: `Job is ${job.status}.` }); | ||
| return; | ||
| } | ||
| await tick(scheduler, job); | ||
| const after = await service.store.get(req.params.id); | ||
| res.json({ ok: true, job: after ? toView(after) : null }); |
| async create(job: NewPollJob): Promise<PollJob> { | ||
| const full: PollJob = { | ||
| ...job, | ||
| id: `job-${++this.seq}`, | ||
| baselineHash: null, | ||
| baselineValues: null, | ||
| lastNotifiedHash: null, | ||
| lastStatus: null, | ||
| lastCheckedAt: null, | ||
| lastTriggeredAt: null, | ||
| triggerCount: 0, | ||
| consecutiveFailures: 0, | ||
| leaseUntil: null, | ||
| createdAt: new Date(job.nextRunAt), | ||
| }; | ||
| this.jobs.set(full.id, full); | ||
| return { ...full }; | ||
| } |
| "typecheck": "tsc --noEmit", | ||
| "test": "vitest run", | ||
| "test:watch": "vitest", | ||
| "test:coverage": "vitest run --coverage" | ||
| "test:coverage": "vitest run --coverage", | ||
| "postinstall": "prisma generate", | ||
| "db:push": "prisma db push" |
| const active = await deps.store.countActiveByOrg(input.orgId); | ||
| if (active >= limits.maxActivePerOrg) { | ||
| throw new ValidationError('orgId', `the organization already has ${active} active polls (cap ${limits.maxActivePerOrg})`); | ||
| } |
…try, atomic org cap, leased forced-run, hardening - Dockerfile: copy prisma/ before npm ci (postinstall runs prisma generate), prisma CLI to prod deps + regenerate after npm prune (prune can drop the generated .prisma client). - poll.expired delivery retries forever (capped backoff) instead of hitting the failure limit and downgrading to poll.failed — core must never permanently miss a terminal event (cor:web:030:03). - Org cap enforced atomically in the store (pg advisory xact lock around count+insert); creation rejects an explicit contentKind that conflicts with the conditions (a doomed job). - POST /polls/:id/run claims the scheduler's lease (claimOne) — 409 poll_leased while the loop holds the job. - startScheduler catches loop errors (no unhandled rejection); forwarder drains response bodies (undici socket reuse); listByOrg capped at 100; InMemory createdAt uses a real clock. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
All review points addressed:
101 tests green, typecheck clean. |
|
To use Codex here, create an environment for this repo. |
P2 of #4: the stateful polling plane, implementing specs
cor:web:030:00..03. The/opsplane is untouched and stays per-call stateless (cor:web:000as rescoped).What's in
Store seam —
PollStoreinterface with two implementations:PrismaPollStore(Postgres, production) andInMemoryPollStore(test suite + DB-less dev), keeping the repo's no-external-deps-in-tests convention. Claiming is lease-based and atomic (UPDATE … WHERE id IN (SELECT … FOR UPDATE SKIP LOCKED) RETURNING) so a second replica is safe by construction. Model isWebFetchPollJob(platform-ordered naming).Credential at rest (
cor:web:030:02) — AES-256-GCM underTOKEN_ENCRYPTION_KEYwith a key-id fingerprint for rotation detection; encrypted at creation, decrypted ONLY in the tick path, never on any read surface, disposed on cancel/expiry/failure/done./pollsroutes — create/get/list/cancel +POST /polls/:id/run(forced tick on the scheduler code path, for e2e). Creation fails fast: interval floor, mandatory TTL ≤ cap, per-org active cap, condition validation, the ops-plane header/auth policy, and the full URL guard runs at creation so a forbidden target never becomes a job. No authorization here — orgId/appId are core-trusted (invariant).Scheduler tick (
cor:web:030:01/:03) — expiry-first (terminal events retried until core accepts; a watch never silently ends), per-tick full guard re-run (nothing trusted across ticks), first-observation baseline init, fire policiesonce/every_change/cooldownwith edge-triggered re-fire, advance-only-after-delivery (baseline/lastNotifiedHash move only on 2xx from core; at-least-once, core dedupes on jobId+snapshotHash+kind), exponential backoff with terminalpoll.failedat the limit (best-effort event — core was down the whole run).Forwarder — outbound-only POST to
CORE_EVENTS_URL(bearerCORE_EVENTS_TOKEN); throws on non-2xx (the gmail forwarder-must-throw invariant). No inbound surface: deployment stays internal-only.Boot — the plane activates only on the full env quorum (
DATABASE_URL+TOKEN_ENCRYPTION_KEY+CORE_EVENTS_URL); otherwise ops-only, unchanged. Prisma 6 (classic client, archetype parity with the other stateful tools);postinstallgenerates so CInpm cikeeps working;npm run db:pushfor deploy.Tests
20 new (98 total green): first-observation init, edge-triggered policies + cooldown, advance-only-after-delivery with core-down retry, backoff curve + failure-limit termination, expiry-accepted-transition + never-fetches, lease claiming, creation caps + guard fail-fast, credential-never-on-a-read-surface, cancel-disposes-credential, plane-absent-when-unconfigured.
Not in (deliberate)
Core wiring is hadron-server#667 (in flight). Deploy (alpha DB + Doppler quorum) is P4, tracked in #4.
🤖 Generated with Claude Code