Server state with PubSub

Fan out server-side state to every connected client with an Effect PubSub and a streaming action.

Server islands hold state in server memory. To push changes to every connected client, publish snapshots to an Effect PubSub and expose one streaming action per module.

The hub pattern

// src/lib/tasks.server.tsx
import * as import EffectEffect from "effect/Effect";
import * as import PubSubPubSub from "effect/PubSub";
import * as import StreamStream from "effect/Stream";
import { function action<Args extends unknown[], Result>(fn: (...args: Args) => Result): typeof fn & ((...args: [...Args, options: {
    signal?: AbortSignal;
}]) => Result)Marks a `*.server.ts` export as a remote RPC action. Runtime identity; the
second call signature adds the transport-only `{ signal }` argument.action } from "oxidejs";

type type Task = {
    id: string;
    text: string;
    completed: boolean;
}Task = { id: stringid: string; text: stringtext: string; completed: booleancompleted: boolean };

const const tasks: Task[]tasks: type Task = {
    id: string;
    text: string;
    completed: boolean;
}Task[] = [];
const const hub: PubSub.PubSub<Task[]>hub = import EffectEffect.const runSync: <PubSub.PubSub<Task[]>, never>(effect: Effect.Effect<PubSub.PubSub<Task[]>, never, never>) => PubSub.PubSub<Task[]>Executes an effect synchronously and returns its success value.

**When to use**

Use when you need to execute an effect that is guaranteed to complete
synchronously.

**Details**

If the effect fails, dies, is interrupted, or performs asynchronous work,
`runSync` throws a `FiberFailure` instead of returning a value. Use
`runSyncExit` when you want the failure captured as an `Exit`.

**Example** (Running a synchronous effect)

```ts import.meta.vitest
import { Effect } from "effect"
const output: Array<unknown> = []

const program = Effect.sync(() => {
  void output.push("Hello, World!")
  return 1
})

const result = Effect.runSync(program)
void output.push(result)
output // => ["Hello, World!", 1]
```

**Example** (Throwing for failed or async effects)

```ts import.meta.vitest
import { Effect } from "effect"
const output: Array<unknown> = []

try {
  // Attempt to run an effect that fails
  Effect.runSync(Effect.fail("my error"))
} catch (e) {
  void output.push("failed effect")
}
try {
  // Attempt to run an effect that involves async work
  Effect.runSync(Effect.promise(() => Promise.resolve(1)))
} catch (e) {
  void output.push("async effect")
}
output // => ["failed effect", "async effect"]
```
@see{@link runSyncExit} for a version that returns an `Exit` type instead of
throwing an error.@categoryrunning@since2.0.0runSync(
  import PubSubPubSub.const unbounded: <Task[]>(options?: {
    readonly replay?: number | undefined;
}) => Effect.Effect<PubSub.PubSub<Task[]>, never, never>Creates an unbounded `PubSub`.

**Example** (Creating an unbounded PubSub)

```ts import.meta.vitest
import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  // Create unbounded PubSub
  const pubsub = yield* PubSub.unbounded<string>()

  const subscription = yield* PubSub.subscribe(pubsub)

  // Can publish unlimited messages
  for (let i = 0; i < 3; i++) {
    yield* PubSub.publish(pubsub, `message-${i}`)
  }

  return yield* PubSub.takeAll(subscription)
}))

const actual = await Effect.runPromise(program)
actual // => ["message-0", "message-1", "message-2"]
```
@categoryconstructors@since2.0.0unbounded<type Task = {
    id: string;
    text: string;
    completed: boolean;
}Task[]>({ replay?: number | undefinedreplay: 1 }),
);
const const notify: () => booleannotify = () =>
  import EffectEffect.const runSync: <boolean, never>(effect: Effect.Effect<boolean, never, never>) => booleanExecutes an effect synchronously and returns its success value.

**When to use**

Use when you need to execute an effect that is guaranteed to complete
synchronously.

**Details**

If the effect fails, dies, is interrupted, or performs asynchronous work,
`runSync` throws a `FiberFailure` instead of returning a value. Use
`runSyncExit` when you want the failure captured as an `Exit`.

**Example** (Running a synchronous effect)

```ts import.meta.vitest
import { Effect } from "effect"
const output: Array<unknown> = []

const program = Effect.sync(() => {
  void output.push("Hello, World!")
  return 1
})

const result = Effect.runSync(program)
void output.push(result)
output // => ["Hello, World!", 1]
```

**Example** (Throwing for failed or async effects)

```ts import.meta.vitest
import { Effect } from "effect"
const output: Array<unknown> = []

try {
  // Attempt to run an effect that fails
  Effect.runSync(Effect.fail("my error"))
} catch (e) {
  void output.push("failed effect")
}
try {
  // Attempt to run an effect that involves async work
  Effect.runSync(Effect.promise(() => Promise.resolve(1)))
} catch (e) {
  void output.push("async effect")
}
output // => ["failed effect", "async effect"]
```
@see{@link runSyncExit} for a version that returns an `Exit` type instead of
throwing an error.@categoryrunning@since2.0.0runSync(import PubSubPubSub.const publish: <Task[]>(self: PubSub.PubSub<Task[]>, value: Task[]) => Effect.Effect<boolean> (+1 overload)Publishes a message to the `PubSub` as an `Effect`, returning whether the
message was accepted.

**When to use**

Use when you need to publish from effectful code and let the configured
PubSub strategy handle surplus messages.

**Details**

The effect succeeds with `false` if the `PubSub` is shut down. If the message
cannot be accepted immediately, the configured strategy decides how surplus
messages are handled.

**Example** (Publishing a message)

```ts import.meta.vitest
import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)

  // Publish a message
  const published = yield* PubSub.publish(pubsub, "Hello World")

  const subscription = yield* PubSub.subscribe(pubsub)

  yield* PubSub.publish(pubsub, "Hello")
  const message = yield* PubSub.take(subscription)
  return { published, message }
}))

const actual = await Effect.runPromise(program)
actual // => { published: true, message: "Hello" }
```
@see{@link publishUnsafe} for a synchronous non-blocking attempt that does not run effectful surplus handling@categorypublishing@since2.0.0publish(const hub: PubSub.PubSub<Task[]>hub, [...const tasks: Task[]tasks]));

// The streaming transport: one async generator per module.
export const const getTasks: (() => AsyncGenerator<Task[], void, any>) & ((options: {
    signal?: AbortSignal;
}) => AsyncGenerator<Task[], void, any>)getTasks = action<[], AsyncGenerator<Task[], void, any>>(fn: () => AsyncGenerator<Task[], void, any>): (() => AsyncGenerator<Task[], void, any>) & ((options: {
    signal?: AbortSignal;
}) => AsyncGenerator<Task[], void, any>)Marks a `*.server.ts` export as a remote RPC action. Runtime identity; the
second call signature adds the transport-only `{ signal }` argument.action(async function* () {
  yield* import StreamStream.const toAsyncIterable: <Task[], never>(self: Stream.Stream<Task[], never, never>) => AsyncIterable<Task[]>Converts a stream to an `AsyncIterable` for `for await...of` consumption.

**Example** (Converting to an async iterable)

```ts import.meta.vitest
import { Stream } from "effect"

const stream = Stream.make(1, 2, 3)

await Array.fromAsync(Stream.toAsyncIterable(stream)) // => [1, 2, 3]
```
@categorydestructors@since3.15.0toAsyncIterable(import StreamStream.const fromPubSub: <Task[]>(pubsub: PubSub.PubSub<Task[]>) => Stream.Stream<Task[], never, never>Creates a stream from a subscription to a `PubSub`.

**Example** (Creating a stream from a subscription to a PubSub)

```ts import.meta.vitest
import { Effect, Fiber, PubSub, Stream } from "effect"

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.unbounded<number>({ replay: 3 })

  const fiber = yield* Stream.fromPubSub(pubsub).pipe(
    Stream.take(3),
    Stream.runCollect,
    Effect.forkChild
  )

  yield* PubSub.publish(pubsub, 1)
  yield* PubSub.publish(pubsub, 2)
  yield* PubSub.publish(pubsub, 3)

  const values = yield* Fiber.join(fiber)
  values // => [1, 2, 3]
})

await Effect.runPromise(program)
```
@categoryconstructors@since2.0.0fromPubSub(const hub: PubSub.PubSub<Task[]>hub));
});

export const const addTask: ((text: string) => Promise<void>) & ((text: string, options: {
    signal?: AbortSignal;
}) => Promise<void>)addTask = action<[text: string], Promise<void>>(fn: (text: string) => Promise<void>): ((text: string) => Promise<void>) & ((text: string, options: {
    signal?: AbortSignal;
}) => Promise<void>)Marks a `*.server.ts` export as a remote RPC action. Runtime identity; the
second call signature adds the transport-only `{ signal }` argument.action(async (text: stringtext: string) => {
  const tasks: Task[]tasks.Array<Task>.push(...items: Task[]): numberAppends new elements to the end of an array, and returns the new length of the array.
@paramitems New elements to add to the array.push({
    id: stringid: var crypto: Crypto[MDN Reference](https://developer.mozilla.org/docs/Web/API/Window/crypto)crypto.Crypto.randomUUID(): `${string}-${string}-${string}-${string}-${string}` (+1 overload)[MDN Reference](https://developer.mozilla.org/docs/Web/API/Crypto/randomUUID)randomUUID(),
    text: stringtext,
    completed: booleancompleted: false,
  });
  const notify: () => booleannotify();
});

Three pieces:

The hub holds the change feed. replay: 1 gives late subscribers the latest snapshot immediately — this is your initial render.
Mutations publish a fresh snapshot after each change.
One streaming action is the transport the router wires to the client.

replay: 1 is what makes SSR work: renderToString takes the first stream value, and the replay buffer delivers it instantly.

// src/li/tasks.server.tsx
export const TaskList = async function TaskList() {
  return Stream.map(
    Stream.fromAsyncIterable(getTasks(), (error: unknown) =>
      error instanceof Error ? error : new Error(String(error)),
    ),
    (list: Task[]) => (
      <ul>
        {list.map((task) => (
          <li key={task.id}>{task.text}</li>
        ))}
      </ul>
    ),
  );
};

The scanner sees Stream.fromAsyncIterable(getTasks(), …) and wires getTasks as the client transport. On the browser, the proxy resumes the same generator over RPC — every publish repaints the island.

Why the transport action stays

getTasks looks redundant — the hub is right there. It is the boundary: the hub lives in server memory, and the browser can only reach it through an RPC generator. Keep one streaming export per server module and call it with Stream.fromAsyncIterable inside each island that needs live updates.

Topic	Guide
Server islands	Server islands
Streams and generators	Streams — paint Effect `Stream` values; `when` for per-emission generators
Triggering frames	Server islands

Server state with PubSub

The hub pattern

Subscribe from an island

Why the transport action stays

Related