Op — Managing Async Operations

Rapid asynchronous operations—such as search autocomplete—can result in out-of-order network responses overwriting newer data. Manually managing cancellation with AbortController and component state adds significant boilerplate.

Op<I, E, A> decouples the asynchronous operation from its execution strategy:

Task Definition: The core asynchronous work is defined once using Op.create.
Execution Strategy: Concurrency models (restartable, exclusive, parallel), retries with backoff, and timeouts are configured via Op.interpret.

Creating an Op

Op.create defines an operation by accepting an asynchronous factory function and an error mapper:

import { Op } from "@nlozgachev/pipelined/core";

interface User { id: string; name: string }
class ApiError extends Error { constructor(public reason: unknown) { super() } }

const fetchUser = Op.create(
  (signal) => (id: string) =>
    fetch(`/users/${id}`, { signal }).then((r) => {
      if (!r.ok) throw new Error(`HTTP ${r.status}`);
      return r.json() as Promise<User>;
    }),
  (error) => new ApiError(error),
);

The factory function receives an AbortSignal and returns a function waiting for input. Pass this signal to cancellable operations (such as fetch).

If the async function is defined elsewhere, pass the signal as an argument:

async function getUserData(id: string, signal: AbortSignal): Promise<User> {
  const response = await fetch(`/users/${id}`, { signal });
  return response.json();
}

const fetchUserOp = Op.create(
  (signal) => (id: string) => getUserData(id, signal),
  (error) => new ApiError(error),
);

Op definitions are lazy blueprints; execution begins only when interpreted and triggered.

Lifting plain async functions with `lift`

If you are writing a quick utility or do not require a typed error channel, you can lift a plain async function using Op.lift. Any rejection is captured as Err<unknown> automatically:

const quickSearch = Op.interpret(
  Op.lift((query: string, signal) =>
    fetch(`/search?q=${query}`, { signal }).then((r) => r.json()),
  ),
  { strategy: "restartable" },
);

The Asynchronous Lifecycle

Four steps take you from a raw, uncancellable fetch to a managed UI flow:

flowchart TD
    A[1. Op.create] -->|Describe the work once| B[2. Op.interpret]
    B -->|Select a concurrency strategy| C[3. manager.subscribe]
    C -->|Listen to loading and outcome states| D[4. manager.run]
    D -->|Execute execution| E{Outcome}
    E -->|Success| Ok[Ok A]
    E -->|Failure| Err[Err E]
    E -->|Cancelled or Dropped| Nil[Nil reason]

Create: Describe the core async task once.
Interpret: Choose a concurrency and lifecycle strategy, returning a manager.
Subscribe: Attach listeners to react to loading states and outcomes.
Run: Trigger the operation with an input value.

// 2. Choose our strategy
const userManager = Op.interpret(fetchUserOp, { strategy: "exclusive" });

// 3. Listen to transitions
userManager.subscribe((state) => {
  if (Op.is.pending(state)) showSpinner();
  if (Op.is.ok(state))      renderProfile(state.value);
  if (Op.is.err(state))     showError(state.error);
});

// 4. Trigger execution
userManager.run("user_123");

The Three Outcomes: Ok, Err, and Nil

Every time you call run(), the invocation eventually settles into one of three outcomes: Ok<A>, Err<E>, or Nil.

Ok and Err represent standard success and failure. Nil is introduced to model situations where an invocation did not complete because the concurrency strategy cancelled, dropped, or bypassed it.

To help you diagnose what happened, the Nil variant carries a precise reason:

"aborted" — abort() was called explicitly on the manager.
"dropped" — the invocation was ignored because the strategy was busy and had no remaining capacity.
"replaced" — a newer run() call executed, cancelling this in-flight invocation.
"evicted" — the invocation was removed from a waiting queue or buffer before it even started executing.

Unpacking outcomes

You can unpack invocation outcomes using direct type guards:

const outcome = await userManager.run("user_123");

if (Op.is.ok(outcome)) {
  console.log("Value:", outcome.value);
} else if (Op.is.err(outcome)) {
  console.error("Error:", outcome.error);
} else {
  console.log("Operation skipped:", outcome.reason); // "aborted", "dropped", etc.
}

Alternatively, you can transform values inside Ok using map, or perform comprehensive case mapping using match or fold:

Op.match({
  ok:  (user) => renderDashboard(user),
  err: (err)  => renderAlert(err),
  nil: (nil)  => console.log(`Skipped: ${nil.reason}`),
})(outcome);

Choosing a Concurrency Strategy

The execution strategy dictates what happens when run() is triggered while an existing operation is already in flight. Choosing the correct strategy ensures that race conditions and duplicate submissions are mathematically impossible.

Use this simple breakdown to select a strategy:

Only run once (e.g. initial page setup) → once
Segment by input key (e.g. independently load different rows) → keyed
Run multiple operations in parallel (e.g. bulk file uploads) → concurrent
Run one-at-a-time:
- Cancel the active one and restart → restartable
- Ignore the new one until the active completes → exclusive
- Queue all calls and run them in order → queue
- Buffer calls, keeping only the latest pending one → buffered
Rate-limit or delay calls:
- Wait for the user to stop typing → debounced
- Fire immediately, then enforce a cooldown → throttled

Concurrency Strategy Reference

Strategy	Description & Common Use Cases
`once`	Fires exactly once. Only the first `run()` executes. All subsequent calls resolve immediately to `DroppedNil`. State is permanently cached once the first call finishes. Use for initial data load or system initialization.
`restartable`	New calls immediately cancel any active in-flight request. Only the latest result ever completes. Use for autocomplete inputs, search boxes, and active tab navigation.
`exclusive`	New calls are ignored while an operation is in flight. The active request always runs to completion. Use for form submissions and payment checkouts.
`queue`	Calls are queued and run sequentially in the order they were submitted. Use for sequential file processing or transactional steps.
`buffered`	Maintains exactly 1 active slot + 1 waiting slot. A new call replaces whatever is in the waiting slot. Use for auto-save pipelines.
`debounced`	Waits for a quiet period of N milliseconds before starting. Resets on every new call. Use for live validations and window resizing.
`throttled`	Fires immediately on the first call, then ignores new calls for N milliseconds. Supports an optional trailing edge. Use for scroll listeners and rate-limited actions.
`concurrent`	Runs up to N operations in parallel. If all slots are full, you can choose to `"queue"` or `"drop"` subsequent requests. Use for bounded uploaders and connection pools.
`keyed`	Multiplexes state per input key, running keys in parallel while applying a local sub-strategy (like `restartable` or `exclusive`) to identical keys. Use for per-row dashboard loading.

In-Depth Concurrency Mechanics

Cooldowns with `throttled`

throttled fires immediately on the first leading-edge call and locks out subsequent executions for a cooldown window. By adding trailing: true, you can ensure that the last call made during the cooldown period fires once at the trailing edge of the window:

import { Duration } from "@nlozgachev/pipelined/composition";

const handleResize = Op.interpret(computeLayoutOp, {
  strategy: "throttled",
  duration: Duration.milliseconds(100),
  trailing: true,
});

This guarantees an immediate initial visual response and a final layout calculation once the user stops resizing.

Parallelism with `concurrent`

concurrent runs up to n operations simultaneously. When the capacity is exhausted, the overflow policy determines what happens:

// Queue up to 10 file uploads in order
const uploadManager = Op.interpret(uploadFileOp, {
  strategy: "concurrent",
  n: 3,
  overflow: "queue",
});

With overflow: "queue", subsequent callers see a Queued state carrying a position number. With overflow: "drop", excess requests are rejected with DroppedNil immediately.

Segmented execution with `keyed`

keyed manages an independent state machine for every key extracted from your input.

Unlike other strategies that hold a single state, manager.state on a keyed manager is a ReadonlyMap of keys to states, and the subscriber receives a fresh map snapshot on every state change:

const userProfileManager = Op.interpret(fetchUserOp, {
  strategy: "keyed",
  key: (user) => user.id,
  perKey: "exclusive",
});

userProfileManager.subscribe((map) => {
  for (const [userId, state] of map) {
    if (Op.is.pending(state)) showRowSpinner(userId);
    if (Op.is.ok(state))      renderRow(userId, state.value);
  }
});

userProfileManager.run({ id: "user_a" }); // Starts slot 'user_a'
userProfileManager.run({ id: "user_b" }); // Starts slot 'user_b' in parallel

State Subscriptions and UI Integration

The manager returned by Op.interpret maintains the active state machine. You can subscribe to it to drive UI updates or synchronously read the current state:

const manager = Op.interpret(searchUsers, { strategy: "restartable" });

manager.subscribe((state) => {
  if (Op.is.pending(state)) showSpinner();
  if (Op.is.ok(state))      renderList(state.value);
  if (Op.is.nil(state))     clearUI();
});

The manager.state property is always available for synchronous reads, making it simple to bind to UI frameworks:

// React Integration
const state = useSyncExternalStore(
  (onStoreChange) => manager.subscribe(onStoreChange),
  () => manager.state,
);

To return a manager to Idle without cancelling active in-flight requests, use manager.reset(). To trigger repeated executions on a timed loop, use manager.poll(input, { interval }).

Per-Invocation Awaiting

Every time you invoke manager.run(input), it returns a Deferred value that resolves to the specific outcome of that invocation. This allows you to await outcomes inline inside sequential, imperative flows:

const outcome = await checkoutManager.run(cartData);

if (Op.is.ok(outcome)) {
  navigateTo("/receipt");
} else if (Op.is.err(outcome)) {
  renderAlert(outcome.error.message);
}

You can coordinate multiple runs in parallel using Op.all or Op.race without managing manual promise plumbing:

const [profile, settings] = await Op.all([
  fetchProfile.run(userId),
  fetchSettings.run(userId),
]);

Retry and Timeout Policies

Retries and timeouts are configured once at interpretation time, removing operational noise from your core async actions.

const paymentManager = Op.interpret(submitPaymentOp, {
  strategy: "exclusive",
  retry: {
    attempts: 3,
    backoff: (attempt) => Duration.milliseconds(attempt * 500),
    when: (error) => error.isNetworkTimeout,
  },
  timeout: {
    duration: Duration.seconds(10),
    onTimeout: () => new ApiError("Transaction timed out"),
  },
});

If a retry policy is active, the manager’s state automatically expands to include a Retrying state between attempts, enabling you to display “Retrying lookup (attempt 2)…” cleanly in the UI.

Outcome Conversions

At boundaries, you can map an Outcome back to other library primitives.

Op.to.Result maps Ok to Ok and Err to Err. Because Result has no concept of cancellations, you must provide a fallback error to represent Nil states:

const resultOutcome = pipe(
  outcome,
  Op.to.Result(() => new ApiError("Request was cancelled")),
); // Result<ApiError, A>

Op.to.Maybe maps Ok to Some and maps both Err and Nil states to None:

const maybeOutcome = Op.to.Maybe(outcome); // Some(A) or None

To wire the success of one manager directly to the invocation of another, use Op.wire:

const stopWire = Op.wire(searchManager, (users) => loggingManager.run(users));

Problems it solves

Search-as-you-type and tab navigation without network race conditions (restartable): In autocomplete inputs or rapid tab navigation, fast user actions fire multiple overlapping HTTP requests. Slower earlier responses often arrive after newer ones, overwriting fresh data with stale results. Setting strategy: "restartable" automatically aborts previous in-flight HTTP requests via AbortSignal, freeing up browser socket connections and guaranteeing only the latest network response updates state.
Preventing duplicate mutation requests and double charges (exclusive): Rapidly clicking payment, checkout, or delete buttons can send duplicate HTTP POST/PUT requests to backend servers. Setting strategy: "exclusive" ignores subsequent triggers while a network mutation is actively in flight, structurally eliminating duplicate API submissions without manual component boolean flags.
Auto-saving editor drafts without API congestion (buffered and debounced): In rich text editors and settings forms, sending an HTTP PUT/PATCH request on every single keystroke floods the backend server, while saving only on unmount risks data loss. buffered keeps at most one active network request in flight and one waiting draft (automatically replacing superseded payloads), while debounced waits for typing pauses before dispatching the HTTP save.
Per-entity independent network requests in tables and feeds (keyed): In data tables or dashboard lists where users trigger actions on individual items (such as archiving a row, fetching order details, or retrying a failed webhook), tracking loading and error states per row manually requires complex ID-mapped state dictionaries. keyed multiplexes independent request managers per entity ID, isolating in-flight AbortSignals and state transitions so one item’s slow network request never blocks or cancels another.
Rate-limited batch uploads and connection pool management (concurrent): Firing dozens of simultaneous HTTP requests (such as uploading file batches or fetching paginated collections) can saturate browser connection limits (such as HTTP/1.1 six-connection host limits) or trigger HTTP 429 Too Many Requests errors. concurrent with overflow: "queue" caps simultaneous active requests (e.g. n: 3) and queues the remainder, while overflow: "drop" discards excess requests when rate limits are saturated.
Sequential transactional API queues (queue): In multi-step backend operations (such as creating an account, provisioning tenant databases, and sending invitations) or offline sync queues, HTTP requests must execute in strict chronological order where each request depends on the completion of the previous one. queue serializes all network invocations in strict FIFO order, aborting on failure without manual promise queue plumbing.
Rate-limited telemetry and high-frequency API sync (throttled): High-frequency actions (such as sending real-time cursor presence beacons, analytics pings, or rapid refresh requests) risk overwhelming API endpoints. throttled guarantees network requests fire at most once per time window, executing immediately on the leading edge and dispatching the latest payload on the trailing edge.
One-time session bootstrap and configuration fetching (once): In web applications, fetching auth session tokens, remote feature flags, or global tenant settings during app startup must hit the network exactly once. once guarantees that only the first request reaches the network, permanently caching the result and resolving subsequent calls immediately without repeated API fetches.
Network resilience with declarative retries, timeouts, and polling (retry, timeout, poll): Flaky mobile connections, temporary 503 Service Unavailable spikes, and hanging socket connections require backoff retry strategies and hard timeout limits. Op encapsulates exponential backoff schedules, jitter, HTTP status filter predicates (when: (err) => err.status >= 500), timeout aborts, and automated interval polling (manager.poll) into a declarative configuration blueprint, cleanly decoupled from business code.
Tying request lifecycle states to reactive UI stores without inconsistent flags: Managing network lifecycles with separate isLoading, isError, error, and data flags leads to contradictory UI states (such as spinners showing alongside error alerts). Op managers provide a single reactive state machine with synchronous manager.state reads, making integration with React (useSyncExternalStore), Vue, or Svelte predictable and exhaustive.