Effect Concurrency Testing

Effect Concurrency Testing Skill

This skill provides patterns for testing Effect's concurrency primitives: fibers, latches, deferreds, PubSub, SubscriptionRef, and streams.

Core Principles

CRITICAL: Choose the correct coordination primitive based on what you need to synchronize.

For Stream.fromPubSub subscription registration, prefer an explicit readiness signal when you control the stream. If the API offers no readiness hook and you only need registration to settle before publishing, a tiny Effect.yieldNow() or very short sleep is an acceptable last resort. Avoid broad polling or arbitrary delays.

Fiber Coordination Patterns

Effect.yieldNow - Simple Fiber Scheduling

Use Effect.yieldNow when you need to allow other fibers to execute. This is preferred over TestClock.adjust for non-time-dependent code.

import { it } from '@effect/vitest';
import { Effect, Exit, Fiber, Latch } from 'effect';

it.effect('fiber polling with yieldNow', () =>
	Effect.gen(function* () {
		const latch = yield* Latch.make();

		const fiber = yield* latch.await.pipe(Effect.forkChild);

		yield* Effect.yieldNow();

		expect(fiber.pollUnsafe()).toBeUndefined();

		yield* latch.open;

		expect(yield* Fiber.await(fiber)).toEqual(Exit.void);
	})
);

Latch - Explicit Coordination

Latch.make() creates a gate that blocks fibers until opened:

import { it } from '@effect/vitest';
import { Effect, Fiber, Latch } from 'effect';

it.effect('latch coordination', () =>
	Effect.gen(function* () {
		const latch = yield* Latch.make();

		const fiber = yield* Effect.gen(function* () {
			yield* latch.await;
			return 'completed';
		}).pipe(Effect.forkChild);

		yield* Effect.yieldNow();
		expect(fiber.pollUnsafe()).toBeUndefined();

		yield* latch.open;

		const result = yield* Fiber.join(fiber);
		expect(result).toBe('completed');
	})
);

Latch Operations

import { Latch } from 'effect';

declare const latch: Latch.Latch;

latch.await; // Wait until latch is open
latch.open; // Open the latch (current and future waiters proceed)
latch.close; // Close the latch (future waiters suspend again)
latch.release; // Wake current waiters only; latch stays closed for future waiters
latch.whenOpen; // Run effect only when latch is open

Deferred - Signal Readiness Between Fibers

Use Deferred when one fiber needs to signal another with a value:

import { it } from '@effect/vitest';
import { Effect, Deferred, Fiber } from 'effect';

it.effect('deferred signaling', () =>
	Effect.gen(function* () {
		const signal = yield* Deferred.make<number>();

		const consumer = yield* Effect.gen(function* () {
			const value = yield* Deferred.await(signal);
			return value * 2;
		}).pipe(Effect.forkChild);

		yield* Deferred.succeed(signal, 21);

		const result = yield* Fiber.join(consumer);
		expect(result).toBe(42);
	})
);

fiber.pollUnsafe() - Check Completion Without Blocking

import { Exit, Fiber } from 'effect';

declare const fiber: Fiber.Fiber<string>;

fiber.pollUnsafe();
// Returns undefined if running
// Returns Exit<A, E> if completed (success, failure, or interrupted)

// Check if still running
expect(fiber.pollUnsafe()).toBeUndefined();

// Check if completed
expect(fiber.pollUnsafe()).toBeDefined();

// Check specific completion
expect(fiber.pollUnsafe()).toEqual(Exit.succeed('result'));

PubSub Event Testing

Direct Event Verification

Use Effect.scoped to manage PubSub subscription lifecycle:

import { it } from '@effect/vitest';
import { Effect, PubSub } from 'effect';

it.effect('verify published events', () =>
	Effect.gen(function* () {
		const pubsub = yield* PubSub.unbounded<string>();

		yield* Effect.scoped(
			Effect.gen(function* () {
				const sub = yield* PubSub.subscribe(pubsub);

				yield* PubSub.publish(pubsub, 'event-1');
				yield* PubSub.publish(pubsub, 'event-2');

				const events = yield* PubSub.takeAll(sub);

				expect(events).toEqual(['event-1', 'event-2']);
			})
		);
	})
);

Both events are published before PubSub.takeAll, so it returns immediately. PubSub.takeAll suspends when the subscription is empty and always returns a NonEmptyArray; for non-blocking drains or “no more events” assertions, use PubSub.takeUpTo(sub, n) instead, which returns whatever is buffered (possibly an empty array).

Testing Event Publishers

When testing a service that publishes events:

import { it } from '@effect/vitest';
import { Effect, PubSub, Context, Layer } from 'effect';

interface UserEvent {
	readonly type: 'created' | 'deleted';
	readonly userId: string;
}

class EventBus extends Context.Service<EventBus, PubSub.PubSub<UserEvent>>()(
	'EventBus'
) {}

class UserService extends Context.Service<
	UserService,
	{ readonly createUser: (id: string) => Effect.Effect<void> }
>()('UserService') {}

declare const UserServiceLive: Layer.Layer<UserService, never, EventBus>;

it.effect('should publish user created event', () =>
	Effect.gen(function* () {
		const pubsub = yield* PubSub.unbounded<UserEvent>();

		yield* Effect.scoped(
			Effect.gen(function* () {
				const sub = yield* PubSub.subscribe(pubsub);

				const service = yield* UserService;
				yield* service.createUser('user-123');

				const events = yield* PubSub.takeAll(sub);

				expect(events).toHaveLength(1);
				expect(events[0]).toEqual({
					type: 'created',
					userId: 'user-123'
				});
			}).pipe(
				Effect.provide(UserServiceLive),
				Effect.provide(Layer.succeed(EventBus, pubsub))
			)
		);
	})
);

Concurrent Publisher/Subscriber Testing

import { it } from '@effect/vitest';
import { Deferred, Effect, PubSub, Fiber, Latch, Array as A } from 'effect';

it.effect('concurrent publishers and subscribers', () =>
	Effect.gen(function* () {
		const values = A.range(0, 9);
		const latch = yield* Latch.make();
		const ready = yield* Deferred.make<void>();
		const pubsub = yield* PubSub.bounded<number>(10);

		const subscriber = yield* PubSub.subscribe(pubsub).pipe(
			Effect.flatMap((sub) =>
				Effect.gen(function* () {
					// Signal that the subscription is registered before publishing
					yield* Deferred.succeed(ready, undefined);
					yield* latch.await;
					return yield* Effect.forEach(values, () =>
						PubSub.take(sub)
					);
				})
			),
			Effect.scoped,
			Effect.forkScoped
		);

		// Wait until the subscriber has actually subscribed (PubSub is not
		// an event log — publishing before subscription would drop messages)
		yield* Deferred.await(ready);

		yield* PubSub.publishAll(pubsub, values);
		yield* latch.open;

		const result = yield* Fiber.join(subscriber);
		expect(result).toEqual(values);
	})
);

SubscriptionRef Testing

Testing Stream Changes with Latches

The latch pattern ensures the stream subscription is ready before mutations:

import { it } from '@effect/vitest';
import { Effect, Fiber, Latch, Number } from 'effect';
import { Stream, SubscriptionRef } from 'effect';

it.effect('multiple subscribers can receive changes', () =>
	Effect.gen(function* () {
		const ref = yield* SubscriptionRef.make(0);
		const latch1 = yield* Latch.make();
		const latch2 = yield* Latch.make();

		const fiber1 = yield* SubscriptionRef.changes(ref).pipe(
			Stream.tap(() => latch1.open),
			Stream.take(3),
			Stream.runCollect,
			Effect.forkScoped
		);

		yield* latch1.await;
		yield* SubscriptionRef.update(ref, Number.increment);

		const fiber2 = yield* SubscriptionRef.changes(ref).pipe(
			Stream.tap(() => latch2.open),
			Stream.take(2),
			Stream.runCollect,
			Effect.forkScoped
		);

		yield* latch2.await;
		yield* SubscriptionRef.update(ref, Number.increment);

		const result1 = yield* Fiber.join(fiber1);
		const result2 = yield* Fiber.join(fiber2);

		expect(result1).toEqual([0, 1, 2]);
		expect(result2).toEqual([1, 2]);
	})
);

Testing Subscription Interruption

import { it } from '@effect/vitest';
import { Effect, Exit, Fiber, Latch, Number } from 'effect';
import { Pull, Stream, SubscriptionRef } from 'effect';

it.effect('subscriptions are interruptible', () =>
	Effect.gen(function* () {
		const ref = yield* SubscriptionRef.make(0);
		const latch = yield* Latch.make();

		const fiber = yield* SubscriptionRef.changes(ref).pipe(
			Stream.tap(() => latch.open),
			Stream.take(10),
			Stream.runCollect,
			Effect.forkScoped
		);

		yield* latch.await;
		yield* SubscriptionRef.update(ref, Number.increment);
		yield* Fiber.interrupt(fiber);

		const result = yield* Fiber.await(fiber);

		expect(Exit.hasInterrupts(result)).toBe(true);
	})
);

Stream Testing

Collecting Stream Results

import { it } from '@effect/vitest';
import { Effect } from 'effect';
import { Stream } from 'effect';

it.effect('should collect stream elements', () =>
	Effect.gen(function* () {
		const result = yield* Stream.make(1, 2, 3, 4, 5).pipe(
			Stream.filter((n) => n % 2 === 0),
			Stream.runCollect
		);

		expect(result).toEqual([2, 4]);
	})
);

Testing Stream Side Effects

import { it } from '@effect/vitest';
import { Effect, Ref } from 'effect';
import { Stream } from 'effect';

it.effect('should track side effects', () =>
	Effect.gen(function* () {
		const log = yield* Ref.make<string[]>([]);

		yield* Stream.make('a', 'b', 'c').pipe(
			Stream.tap((item) => Ref.update(log, (items) => [...items, item])),
			Stream.runDrain
		);

		const logged = yield* Ref.get(log);
		expect(logged).toEqual(['a', 'b', 'c']);
	})
);

Testing Stream Errors

import { it } from '@effect/vitest';
import { Effect, Exit, Schema } from 'effect';
import { Stream } from 'effect';

class StreamError extends Schema.TaggedError<StreamError>()(
	'StreamError',
	{
		message: Schema.String
	}
) {}

it.effect('should handle stream errors', () =>
	Effect.gen(function* () {
		const result = yield* Stream.make(1, 2, 3).pipe(
			Stream.mapEffect((n) =>
				n === 2
					? Effect.fail(new StreamError({ message: 'boom' }))
					: Effect.succeed(n)
			),
			Stream.runCollect,
			Effect.exit
		);

		expect(Exit.isFailure(result)).toBe(true);
	})
);

Testing Stream Finalization

import { it } from '@effect/vitest';
import { Effect, Ref } from 'effect';
import { Stream } from 'effect';

it.effect('should run finalizers', () =>
	Effect.gen(function* () {
		const finalized = yield* Ref.make(false);

		yield* Stream.make(1, 2, 3).pipe(
			Stream.ensuring(Ref.set(finalized, true)),
			Stream.take(1),
			Stream.runDrain
		);

		expect(yield* Ref.get(finalized)).toBe(true);
	})
);

Interruption Testing

Testing Fiber Interruption

import { it } from '@effect/vitest';
import { Effect, Exit, Fiber, Cause } from 'effect';

it.effect('should handle interruption', () =>
	Effect.gen(function* () {
		const fiber = yield* Effect.never.pipe(Effect.forkChild);

		yield* Fiber.interrupt(fiber);

		const result = yield* Fiber.await(fiber);

		expect(Exit.hasInterrupts(result)).toBe(true);
	})
);

Testing Interrupted-Only Cause

import { it } from '@effect/vitest';
import { Effect, Exit, Fiber, Cause } from 'effect';

it.effect('should have interrupted-only cause', () =>
	Effect.gen(function* () {
		const fiber = yield* Effect.never.pipe(Effect.forkChild);

		yield* Fiber.interrupt(fiber);

		const result = yield* Fiber.await(fiber);

		expect(
			Exit.isFailure(result) && Cause.hasInterruptsOnly(result.cause)
		).toBe(true);
	})
);

Time-Dependent Concurrency Testing

Use TestClock only when testing time-dependent behavior like delays, timeouts, or schedules.

import { it } from '@effect/vitest';
import { Effect, Fiber, Duration } from 'effect';
import { TestClock } from 'effect/testing';

it.effect('should handle delayed concurrent operations', () =>
	Effect.gen(function* () {
		const fiber = yield* Effect.gen(function* () {
			yield* Effect.sleep(Duration.seconds(5));
			return 'done';
		}).pipe(Effect.forkChild);

		yield* TestClock.adjust(Duration.seconds(5));

		const result = yield* Fiber.join(fiber);
		expect(result).toBe('done');
	})
);

Anti-Patterns

DON'T use TestClock for non-time-dependent code

import { Effect, Duration } from 'effect';
import { TestClock } from 'effect/testing';

// BAD - Using TestClock when not needed
Effect.gen(function* () {
	const fiber = yield* someEffect.pipe(Effect.forkChild);
	yield* TestClock.adjust(Duration.millis(100));
	yield* Fiber.join(fiber);
});

// GOOD - Use yieldNow for simple yielding
Effect.gen(function* () {
	const fiber = yield* someEffect.pipe(Effect.forkChild);
	yield* Effect.yieldNow();
	yield* Fiber.join(fiber);
});

DON'T poll in a loop without yieldNow

import { Effect, Fiber } from 'effect';

declare const fiber: Fiber.Fiber<void>;

// BAD - Busy loop
while (fiber.pollUnsafe() === undefined) {
	// Spins forever!
}

// GOOD - Yield between polls or use Fiber.await
Effect.gen(function* () {
	while (fiber.pollUnsafe() === undefined) {
		yield* Effect.yieldNow();
	}
});

// BETTER - Just await the fiber
Effect.gen(function* () {
	yield* Fiber.await(fiber);
});

DON'T forget Effect.scoped for PubSub subscriptions

import { Effect, PubSub } from 'effect';

declare const pubsub: PubSub.PubSub<string>;

// BAD - Subscription leaks
Effect.gen(function* () {
	const sub = yield* PubSub.subscribe(pubsub);
	// Sub is never cleaned up!
});

// GOOD - Scoped subscription
Effect.gen(function* () {
	yield* Effect.scoped(
		Effect.gen(function* () {
			const sub = yield* PubSub.subscribe(pubsub);
			// takeUpTo never suspends; takeAll would hang here (nothing published)
			const events = yield* PubSub.takeUpTo(sub, 10);
			// Sub cleaned up when scope closes
		})
	);
});

DON'T start subscriptions after mutations

import { Effect, Fiber } from 'effect';
import { Stream, SubscriptionRef } from 'effect';

declare const ref: SubscriptionRef.SubscriptionRef<number>;

// BAD - May miss events: the subscription starts AFTER the first mutation
Effect.gen(function* () {
	yield* SubscriptionRef.update(ref, (n) => n + 1);

	const fiber = yield* SubscriptionRef.changes(ref).pipe(
		Stream.take(1),
		Stream.runCollect,
		Effect.forkChild
	);

	yield* SubscriptionRef.update(ref, (n) => n + 1);

	const result = yield* Fiber.join(fiber);
});

Quality Checklist

• Using correct coordination primitive for the use case
• Effect.scoped wraps PubSub subscriptions
• Latches ensure stream subscriptions are ready before mutations
• Effect.yieldNow used instead of TestClock for non-time-dependent code
• Fiber interruption tested with Exit.hasInterrupts or Cause.hasInterruptsOnly
• Stream finalizers verified with Stream.ensuring
• No busy polling without yields
• Test is deterministic (no race conditions)