Example: an Agent with a schedule and WebSocket connections

A live poll is kept by an Agent, built on the Cloudflare Agents SDK. Voters connect over WebSockets through the SDK's router and cast a vote; the Agent checks each voter against a membership API, counts the vote in its state and tells every connection the new tally. The first vote schedules the poll to close a minute later, when the Agent writes the final tally into a table of its own.

Run the check

From examples/agent-poll in the repository:

pnpm install
pnpm check

The scenario in poll.sim.ts opens two connections and sends a vote on each, at once. The Agent runs as written, over real WebSocket connections, with its state in SQLite and its schedule over the Durable Object's alarm, which fires in virtual time, so the minute takes no time. There are no injected faults; two votes in flight are enough. The check reports:

VIOLATED  poll  (seed poll)

Invariant result-counts-every-vote is violated: 2 row(s) of p-1/results in DO:POLL violate the condition: {"votes":1}, {"votes":1}

Schedule (run seed ...), events kept: ann, bob, ann:1, bob:1
  no injected faults

Each vote read the tally, then awaited the membership check. An Agent is a Durable Object, whose input gate holds other events while it waits on its own storage but not while it awaits an outbound fetch, so the second vote ran meanwhile and read the same tally of 0. Both votes wrote 1, and both scheduled a close: the poll closed twice, and each result counts one vote where two came in. A test that sends one vote at a time never sees it.

The fix

worker.fixed.ts checks membership first, then reads and writes the tally with no await between them, so a vote that arrives meanwhile reads the tally already taken, and only the first vote schedules the close. pnpm check:fixed reports BOUNDED with two votes in flight.

Using it in your own project

websocket([...]) opens connections through your Worker's fetch and sends each connection's messages into the object that accepted it; maxConcurrentInvocations above 1 lets them interleave at the object's I/O. durableObjectTable reads a table of the Agent's own storage at the end of the run.

The code

The complete project is examples/agent-poll in the repository.

worker.ts

/**
 * A live poll kept by an Agent (Cloudflare Agents SDK). Voters connect over
 * WebSockets and cast a vote; each voter is checked against a membership
 * API, the vote is counted in the Agent's state, and every connection is
 * told the new tally. The first vote schedules the poll to close a minute
 * later, when the Agent writes the result into its own table. Run `eutaxy
 * check ./poll.sim.ts` to see two votes arrive at once.
 */

import { Agent, type Connection, routeAgentRequest, type WSMessage } from "agents";

export interface Env {
  readonly POLL: DurableObjectNamespace<Poll>;
}

export interface PollState {
  readonly votes: number;
  readonly closed: boolean;
}

export interface Vote {
  readonly voter: string;
}

export class Poll extends Agent<Env, PollState> {
  override initialState: PollState = { votes: 0, closed: false };

  override async onConnect(connection: Connection): Promise<void> {
    connection.send(JSON.stringify({ type: "tally", votes: this.state.votes }));
  }

  override async onMessage(connection: Connection, message: WSMessage): Promise<void> {
    const vote = JSON.parse(String(message)) as Vote;
    if (this.state.closed) return connection.send(JSON.stringify({ type: "closed" }));
    const votes = this.state.votes;
    if (votes === 0) await this.schedule(60, "close");
    await this.checkMember(vote.voter);
    this.setState({ ...this.state, votes: votes + 1 });
    this.broadcast(JSON.stringify({ type: "tally", votes: votes + 1 }));
  }

  /** The scheduled close: the final tally into the Agent's own table, and to every connection. */
  async close(): Promise<void> {
    this.sql`CREATE TABLE IF NOT EXISTS results (votes INTEGER NOT NULL)`;
    this.sql`INSERT INTO results (votes) VALUES (${this.state.votes})`;
    this.setState({ ...this.state, closed: true });
    this.broadcast(JSON.stringify({ type: "closed", votes: this.state.votes }));
  }

  protected async checkMember(voter: string): Promise<void> {
    const res = await fetch(`https://members.example/members/${voter}`);
    if (!res.ok) throw new Error(`${voter} may not vote (${res.status})`);
  }
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    return (await routeAgentRequest(request, env)) ?? new Response("not found", { status: 404 });
  },
} satisfies ExportedHandler<Env>;

poll.sim.ts

import {
  defineScenario,
  durableObject,
  durableObjectTable,
  everyRow,
  external,
  websocket,
} from "eutaxy/cloudflare";

export default defineScenario({
  name: "poll",
  worker: { entry: "./worker.ts", config: "./wrangler.jsonc" },
  // The poll p-1 starts empty; the SDK creates its own tables.
  setup: [durableObject("POLL", { "p-1": {} })],
  // Two voters connect through the SDK's router and vote at once.
  drive: websocket([
    { id: "ann", path: "/agents/poll/p-1", messages: [{ data: { voter: "ann" } }] },
    { id: "bob", path: "/agents/poll/p-1", messages: [{ data: { voter: "bob" } }] },
  ]),
  // The membership API answers 200 for every voter.
  externals: [external("members", "members.example")],
  invariants: [
    // The schedule closes the poll a minute after the first vote: its result counts both votes.
    everyRow("result-counts-every-vote", durableObjectTable("POLL", "p-1", "results"), [
      { column: "votes", op: "eq", value: 2 },
    ]),
  ],
  // No faults: two votes in flight at once are enough.
  bounds: { maxFaultsPerRun: 0, maxRedeliveries: 0, maxConcurrentInvocations: 2, budgetMs: 2000 },
});

worker.fixed.ts

/**
 * The same poll, with the membership check before the tally is read: the
 * read and the write of the count have no await between them, so a vote
 * that arrives meanwhile reads the count already taken. `eutaxy check
 * ./poll.fixed.sim.ts` reports BOUNDED.
 */

import type { Connection, WSMessage } from "agents";
import { Poll as PlantedPoll, type Vote } from "./worker.ts";

export { default } from "./worker.ts";

export class Poll extends PlantedPoll {
  override async onMessage(connection: Connection, message: WSMessage): Promise<void> {
    const vote = JSON.parse(String(message)) as Vote;
    await this.checkMember(vote.voter);
    if (this.state.closed) return connection.send(JSON.stringify({ type: "closed" }));
    const votes = this.state.votes + 1;
    this.setState({ ...this.state, votes });
    if (votes === 1) await this.schedule(60, "close");
    this.broadcast(JSON.stringify({ type: "tally", votes }));
  }
}