Durable streams

A NetScript stream is a typed change-log using the Durable Streams protocol: producers write entity state into a durable-stream server, and any number of HTTP/SSE consumers materialize the latest value per key. alpha

NetScript's streams capability is the typed, change-data backbone the other plugins lean on — workers, sagas, and the auth service all publish their live state through it. The producer half is implemented: you define a typed stream schema with defineStreamSchema, open a producer with createDurableStream (or the Service-facing createServiceStreamProducer), and upsert/delete/flush entity state over a durable-stream server that runs as an Aspire resource on its assigned port.

The producer runtime in @netscript/plugin-streams-core writes through @durable-streams/client with idempotent delivery. @netscript/plugin-streams carries the service manifest and defineStreamTopic definitions. Runtime producers use the core package; consumers read the durable-stream server's HTTP/SSE protocol to materialize state.

What it is

A NetScript stream is an entity-oriented change log. You describe a set of collections — each a named entity type with a primary key — and the producer publishes upsert and delete operations keyed by that primary key. Downstream readers materialize the latest value per key and observe a live, replayable view of your domain state. This is the same contracts-first instinct as oRPC services, but applied to state replication instead of request/response: the schema is the type contract that both producer and any HTTP/SSE consumer are locked to.

Streams sit alongside the other long-running capabilities rather than replacing them. Reach for a durable saga

when you need message-driven orchestration with compensation; reach for a

trigger when inbound HTTP or a file-watch should kick off work. A stream is the read-model fan-out: each of those plugins runs a thin streams/producer.ts that mirrors its execution state through createDurableStream, so a Fresh dashboard can watch saga, worker, and trigger progress live without polling a request/response API.

The workflow this replaces is familiar: a dashboard polls GET /executions on an interval, the interval is always wrong — too fast for the server, too slow for the operator watching a stuck import — and every new consumer re-implements the same polling loop against a slightly different endpoint. With a stream, the contract moves to the data itself. The live-dashboard tutorial builds exactly this: worker execution state published once through a producer, read by a Fresh page that updates without a polling loop anywhere in it. And because the schema is a frozen, typed collection map, what a stream carries is enumerable from its definition — inspectStreamTopic renders it as a JSON-stable report, which is as useful to a CLI doctor or a coding agent as it is to a test.

Worker execution progress ordering and replay

Worker handlers report progress through an execution-local outbound channel. Calls are persisted in invocation order for that execution, including repeated percentages and decreasing values; the runtime does not debounce, deduplicate, or coalesce them. A terminal success or failure waits for all accepted progress writes to drain, so an unawaited progress call cannot race past completion. Progress persistence failures fail the execution instead of being reduced to log messages.

Each accepted transition publishes the existing full execution-record upsert. Durable-stream replay therefore remains entity replay: offsets order the retained updates, and reducing by execution id materializes the newest record with its last progress value and terminal status. Intermediate updates may be visible in the retained log, but this is not a separate progress-history API.

For React Native screens, follow Consume durable streams in Expo. It uses the same TanStack DB engine through an injected fetch and includes the native runtime probe.

Learn → / Do →

Minimal example — produce, then consume

The producer side is two calls: freeze a typed schema, open a stream, write entity state. The consumer side is an HTTP/SSE read of the same stream path — there is no in-process subscribe() handle, so a Fresh island (or any SSE client) reads the durable log directly and materializes the latest value per key. This is deliberate: exposing one HTTP/SSE surface instead of an in-process subscribe() means a browser island and a server-side consumer read the durable log through the exact same path and wire format, with no separate in-VM subscription API to keep in sync.

// streams/executions-schema.ts
// Author against the core package — this is the live surface.
import { defineStreamSchema } from '@netscript/plugin-streams-core';
import { z } from 'zod';

// Each collection is an entity type with a primary key. The schema is the
// type contract producers and HTTP/SSE consumers are locked to.
export const executionsSchema = defineStreamSchema({
  execution: {
    schema: z.object({
      id: z.string().min(1),
      status: z.enum(['queued', 'running', 'succeeded', 'failed']),
      updatedAt: z.string().datetime(),
    }),
    type: 'execution',
    primaryKey: 'id',
  },
});

// streams/producer.ts
import { createDurableStream } from '@netscript/plugin-streams-core';
import { executionsSchema } from './executions-schema.ts';

// createDurableStream returns a singleton producer per streamPath and begins
// connecting to the durable-stream server immediately.
const producer = createDurableStream({
  streamPath: '/workers/executions',
  schema: executionsSchema,
  producerId: 'workers-service',
});

// upsert/delete are synchronous enqueues keyed by the collection primary key.
producer.upsert('execution', {
  id: 'exec-1',
  status: 'running',
  updatedAt: new Date().toISOString(),
});
producer.delete('execution', 'exec-0');

// flush before graceful shutdown; it rethrows the connect error if the
// producer never connected (see known limitations).
await producer.flush();

// islands/ExecutionsView.tsx — read the durable log directly
import { buildStreamUrl, getStreamsUrl } from '@netscript/plugin-streams-core';
import { bindStreamEventSourceV1 } from '@netscript/fresh/streams';

// Native EventSource needs the real stream route, an opaque replay offset, and
// live=sse. It cannot attach arbitrary authorization headers, so use this from
// a same-origin/cookie-compatible route or proxy when the service is protected.
const url = new URL(buildStreamUrl('/workers/executions', getStreamsUrl()));
url.searchParams.set('offset', '-1');
url.searchParams.set('live', 'sse');
const source = new EventSource(url);

const latest = new Map<string, unknown>();
const binding = bindStreamEventSourceV1({
  source,
  onEvent(event) {
    // The helper registers named data/control listeners and validates both
    // payload schemas before delivering this exhaustive outcome union.
    if (event.event === 'error') { latest.set('$error', event.payload); return; }
    if (event.event !== 'data') return; // control / derived heartbeat
    for (const change of event.payload) { // data is always an ordered array
      if (change.headers.operation === 'delete') latest.delete(change.key);
      else latest.set(change.key, change.value);
    }
  },
});

// An island should call binding.dispose() from its effect cleanup. This plain
// browser example closes on page teardown; W3-A owns reconnect policy.
addEventListener('pagehide', () => binding.dispose(), { once: true });

Breaking migration in 0.0.8

@netscript/plugin-streams no longer exports defineStreamProducer, defineStreamConsumer, StreamProducerHandle, StreamConsumerHandle, StreamUnsupportedOperationError, or unsupportedStreamOperation. streamsPlugin, defineStreamTopic, StreamTopicDefinition, and StreamPayloadSchema remain available for manifest declarations.

For producers, import defineStreamSchema and createDurableStream (or createServiceStreamProducer in backend services) from @netscript/plugin-streams-core. Convert the topic payload schema to a named collection with a primaryKey, as shown above. Replace publish(payload) with upsert(collection, payload) or delete(collection, key); call flush() and close() during shutdown. This is an entity-state transport: generic message payloads need stable entity ids and an appropriate collection model.

For consumers, replace subscribe(handler) with HTTP/SSE reads of the same stream path. buildStreamUrl and getStreamsUrl from @netscript/plugin-streams-core resolve that URL; bindStreamEventSourceV1 from @netscript/fresh/streams validates named SSE events in browser clients, as shown above. Dispose the binding when the client tears down. There is no in-process topic subscription replacement.

Key types first — the stream definition API

A stream schema is a map of collections. Each collection is a CollectionDefinition — a Standard-Schema validator, a State-Protocol type discriminator, and the primaryKey property the producer keys writes by. defineStreamSchema(collections) freezes that map into a StateSchema that both the producer and any HTTP/SSE consumer are locked to.

defineStreamSchema returns a frozen StateSchema<TDef> — the durable-streams runtime attaches per-collection event helpers (insert/update/upsert/delete, the CollectionEventHelpers) so the schema can both validate and emit State-Protocol ChangeEvents. The supported Operation set is insert | update | delete | upsert.

Producer options — createDurableStream

createDurableStream(options) takes a DurableStreamProducerOptions and returns a DurableStreamProducer. It is a singleton factory keyed by streamPath: calling it twice with the same path returns the same live producer (a closed one is replaced). Writes are idempotent via @durable-streams/client's IdempotentProducer (stable producerId + auto-claim), so duplicate enqueues do not double-apply downstream.

The DurableStreamProducer it returns exposes a small, synchronous-write surface with an async flush/close for shutdown. (StreamProducerPort is the implemented-by interface — the same four members, with entityType widened to string.)

Bound stream retention

Set retention: { kind: 'ttl', ttlSeconds: 604800 } or retention: { kind: 'expires-at', expiresAt: '2030-10-08T00:00:00Z' } on a producer before creating its stream. On the shipped server, TTL is a sliding inactivity window renewed by reads and appends; HEAD and reopening with PUT do not renew it or change the policy. Use absolute expiry or rotating day segments for a hard bound on an active stream. Absolute expiry does not slide. Both policies expire the whole stream. Omit retention for the current unbounded behavior. For existing day segments, headDurableStream reads metadata and deleteDurableStream deletes the whole stream through a versioned administrative port.

Run deletion from background workers or triggers with a service identity, independently of any app session. Entity producer.delete(...) appends a tombstone and does not shrink the log. See the stream retention how-to for server TTL, absolute expiry, scheduled segment deletion, typed failure handling, and document day baselines.

Service-side producers — createServiceStreamProducer

When the writer is a backend Service — for example an ingestion worker that emits a doc.ready completion event from the callback where the work finishes — reach for createServiceStreamProducer instead of wiring createDurableStream and the URL/auth resolvers by hand. It is a thin wrapper over createDurableStream that reuses the exact same singleton producer and Aspire discovery (getStreamsUrl / getStreamsAuth). Stream discovery is resolved at construction, so a Service that lacks the streams plugin reference throws immediately rather than blocking or silently dropping writes.

// services/ingestion/emit-completion.ts
import {
  createServiceStreamProducer,
  defineStreamSchema,
} from "@netscript/plugin-streams-core";
import { z } from "zod";

const completions = defineStreamSchema({
  completion: {
    schema: z.object({ id: z.string().min(1), status: z.string() }),
    type: "completion",
    primaryKey: "id",
  },
});

// This throws now if neither DURABLE_STREAMS_URL nor
// services__streams__http__0 is wired.
const producer = createServiceStreamProducer({
  streamPath: "/support-chat/completions",
  schema: completions,
  producerId: "support-chat-ingestion",
});

producer.upsert("completion", { id: "run-1", status: "done" });
await producer.flush();

The CLI writes the required plugin reference when the streams plugin is installed. If construction reports a missing streams reference, install the streams plugin and run netscript service generate to reconcile the generated configuration. ServiceStreamProducerOptions has the same fields as DurableStreamProducerOptions; the returned DurableStreamProducer is identical, so the write/flush/close surface below applies unchanged.

Runtime & transport — URL and auth resolution

The producer never hardcodes a host. createDurableStream resolves the durable-stream base URL through getStreamsUrl() and the auth header through getStreamsAuth(), both of which read the environment so the same code works under Aspire, in a browser build, or against an explicit override. buildStreamUrl joins a stream path onto that base, and inspectStreamTopic returns a JSON-stable diagnostic report for a schema (handy in tests and CLI doctors).

Resolved stream URLs

When configuring streamPath (e.g. streamPath: '/workers/executions'), the framework automatically prepends STREAMS_URL_PREFIX = '/v1/stream/netscript' inside buildStreamUrl.

Concrete example:

• streamPath: '/workers/executions'
• baseUrl: 'https://streams.example.test'
• Resolved endpoint: 'https://streams.example.test/v1/stream/netscript/workers/executions'

Storage mode and restart operations

The plugin manifest declares memory as the ephemeral default. The detailed /health payload contains checks[name="streams-storage"].storage with mode: "memory" | "file", durable, and probe: "not-applicable" | "passed". Memory may be healthy while explicitly reporting durable: false: all stream history disappears on service restart. “Durable Streams” names the protocol, not a persistence guarantee for memory storage.

For the default resource working directory, create .netscript/data/streams, ignore .netscript/data/ in Git, and set the plugin environment entry to that path. Deployments must mount persistent storage separately. File mode has an executed real DurableStreamProducer write, flush, service-process stop, restart and HTTP replay proof; memory has a restart/data-loss control. The proof is in plugins/streams/services/src/tests/storage-restart_test.ts and supports orderly process restart only, not power-loss durability, replication or multiple concurrent owners.

Retention is unbounded without a configured expiry/deletion policy. Entity deletes append tombstones; they do not compact history. Protocol TTL slides on reads/appends; absolute expiry is fixed. Both expire whole streams. Use fixed expiry or scheduled whole-stream/day-segment deletion from workers/triggers under service identity for a hard bound; monitor disk space. There is no automatic compaction or archival.

Recovery stops at the last complete native frame, so an incomplete final append can be lost. Missing segment files can read empty; corrupt LMDB metadata may refuse startup. The startup probe verifies directory I/O, not historical integrity or repair. Preserve damaged files and restore a known-good backup if recovery fails.

Stop producers and streams before copying the entire data directory (metadata and segments) for backup. Restore the complete copy while streams is stopped, retain compatible storage/protocol versions, then verify replay before restarting producers. Live file copies are not a supported consistent backup. One process owns a directory. Durable-by-default provisioning and migration remain the 0.0.9 enhancement #2114.

Known limitations

Be deliberate about what the alpha producer does and does not guarantee.

Endpoints & manifest

The streams plugin is registered as a utility/infra plugin — note it requires neither a database nor KV (requiresDb=false, requiresKv=false), unlike workers, sagas, and triggers. Its durable-stream service listens on its assigned port and is wired into the Aspire resource graph so workers, sagas, and auth can publish through it. The port is overridable via STREAMS_PORT or PORT.

The plugin is referenced from netscript.config.ts as ./plugins/streams/mod.ts. Because workers, sagas, and triggers each list streams in their dependencies, it is installed first in the dependency graph and its service comes up so dependent producers have somewhere to write.

Production notes

Kept in sync, compared

Convex is the reference point for live client state: sync there is a property of its hosted database — client queries are reactive subscriptions, and the platform re-runs them when the underlying data changes. A NetScript stream scopes the same promise differently: sync is a property of a declared contract, not of the database. You freeze a schema of collections, a producer writes latest-state per key to the durable-stream service in your own Aspire graph, and any HTTP/SSE consumer materializes that view — independent of which store actually holds your data (the plugin itself requires neither a database nor KV). The practical consequence: you choose per-collection what becomes live state, and everything outside that schema stays on the request/response and persistence paths you already have.

Reference

This hub is intentionally thin — the full generated API lives in the reference.

streams