Skip to content

ARCHITECTURE.md

Latest commit

 

History

History
389 lines (312 loc) · 20.6 KB

ARCHITECTURE.md

File metadata and controls

389 lines (312 loc) · 20.6 KB
Raw

Durable Streams as a Portable Effect App — Services & Layers across Substrates

A companion to EFFECT_REBUILD_PROPOSAL.md. That doc rebuilt the server on Workers + Durable Objects. This doc factors the same Effect app so the protocol core is substrate-agnostic and each platform (Workers+DO, Node+filesystem, Lambda+S3/DDB, in-memory tests) is a swappable set of Layers — without the core changing.

Vocabulary (per the Effect docs — Services, Layers): a Service is a Context.Tag (a unique id + an interface). A Layer is the blueprint that builds a Service (and declares its own dependencies). The key fact this whole design rests on: one Service can have many Layers — Effect's own DatabaseLive vs DatabaseTest example — "platform- or environment-specific variations of identical services." There is no "port/adapter" concept in Effect; that was my bad import. Everything here is just Services and Layers.

1. The reframe: a Durable Object is a per-stream serialized actor

The protocol's unit of work is one stream. Everything hard about it reduces to three things the runtime must provide per stream key:

Concern What it means DO provides it via Off-DO you must build it
Commit atomicity one append's read→validate→commit can't interleave with another's (no two appends claim the same offset, no torn last_stream_seq) single-threaded instance a per-stream mutex, or a DB transaction / CAS
Notification wake the long-poll/SSE readers on this stream the instant it's appended in-memory state shared across the instance's requests polling, or an external bus
Addressing + lifecycle every request for stream X reaches the one context that owns X; created on first use, evicted when idle idFromName(path) + hibernation a keyed registry of scoped contexts

A DO donates all three. That donation is its entire value — not the SQLite. So the design names each as a Service, lets the DO satisfy it with a trivial Layer, and lets other substrates satisfy it with a real (but generic) Layer.

2. Two kinds of Service — the whole modularity story

Every box below is a Service (Context.Tag). The only thing that varies is how many Layers implement it:

flowchart TD
  subgraph fixed["CORE SERVICES — one canonical Layer (the product, written once)"]
    SVC["StreamService — create/append/read/head/delete/fork/runExpiry"]
    CFG["StreamConfig"]
    HTTP["http router + error-map + render  (an HttpApp value, not a Service)"]
    DOM["domain: offset/json/producer/cursor/sse  (pure, not Services)"]
  end
  subgraph swap["SUBSTRATE SERVICES — same Tag, one Layer PER platform"]
    STORE["StreamStore"]
    HUB["StreamHub"]
    SCHED["ExpiryScheduler"]
    PEER["ForkPeer"]
    MUTEX["StreamMutex"]
    REG["StreamRegistry"]
  end
  HTTP --> SVC
  SVC --> STORE & HUB & SCHED & PEER & MUTEX
  SVC --> CFG & DOM
  HTTP --> REG
Loading
  • Core Services (StreamService, StreamConfig) ship one Layer — StreamService.Default, StreamConfig.Default. This is the protocol; it never branches by platform.
  • Substrate Services (StreamStore, StreamHub, ExpiryScheduler, ForkPeer, StreamMutex, StreamRegistry) ship one Layer per platform — StreamStoreSqlLive on Workers, StreamStoreMemoryLive in tests, StreamStoreFsLive on Node. Same Tag, same interface, different Layer.
  • A platform = one composite Layer that merges the substrate-Service Layers for that target: CloudflareLive, NodeLive, MemoryLive. (This is what I sloppily called a "profile.")

StreamService's requirements channel (R) is exactly the six substrate Services + StreamConfig + Clock + Random. That R is the contract a platform must satisfy.

3. The substrate Services

Service Responsibility (the question it answers) Interface sketch
StreamStore "persist + read this stream's meta, chunks, producer state, atomically per commit" readMeta, messagesAfter, getProducer, commit(mutation)
StreamMutex "run this stream's mutation critical section without interleaving" (see §4) serialize(effect)
StreamHub "wake the live readers waiting on this stream" publish(signal), signals: Stream, subscribe: scoped Dequeue
ExpiryScheduler "fire a callback at a future time for TTL cleanup" scheduleAt(epochMs), cancel
ForkPeer "read + ref-count another stream so this one can fork it" exportFrom(srcKey, req), release(srcKey)
StreamRegistry "give me the per-stream context for key K, created on demand, evicted when idle" (see §5) withStream(key, effect)

Clock and Random are Effect built-ins (already swappable: TestClock, seeded Random). Telemetry (Tracer + Logger) is a cross-cutting Layer (Otlp on the edge).

StreamStore exposes an explicit atomic commit(mutation) rather than leaking fine-grained writes — so each substrate satisfies atomicity its own way (DO write-coalescing / Postgres tx / DDB TransactWriteItems / FS journal) instead of relying on the DO's incidental behavior.

4. StreamMutex — what it is, and what it is not (the correction)

It is not "one writer per stream." The Durable Streams protocol is explicitly multi-producer:

  • Stream-Seq is a per-stream optimistic-ordering token — many writers, each proposing a monotonic seq, rejected with 409 if they regress.
  • Producer Id/Epoch/Seq is Kafka-style fencing so several idempotent producers can publish to the same stream concurrently and dedupe independently.

So clients run as many concurrent producers as they like. What must not interleave is the server-side critical section of a single append: read current offset + read producer/seq state → validate (content-type, fencing, seq) → append bytes → advance offset + write producer/seq state. Two appends executing that read-modify-write concurrently could claim the same offset or clobber last_stream_seq. StreamMutex.serialize(effect) makes that section atomic.

  • On a Durable Object: a per-instance Semaphore(1) — same as Node. The DO's JS thread is single-threaded, but the Effect runtime is not interleaving-free: fibers yield to the scheduler every Scheduler.MaxOpsBeforeYield (2048) ops and at any async boundary, and the DO input gate only holds new events out during storage awaits — so two in-flight requests' fibers can interleave mid-append (e.g. a large body split across many chunk inserts). An earlier draft used the identity Layer, leaning on the single-threaded-runtime semantic; under Effect that assumption is unsound, and the semaphore costs nothing when uncontended.
  • On a multi-fiber host (Node): a per-stream Effect.Semaphore(1) — serialize: (e) => sem.withPermits(1)(e).
  • On a shared database / multi-instance: a transaction (SELECT … FOR UPDATE, advisory lock) or an optimistic compare-and-swap on the offset/version with bounded retry — no literal lock at all. (Stream-Seq is the protocol-level expression of that same optimistic idea.)

StreamMutex and the protocol fencing are complementary: fencing decides whether an append is accepted; the mutex ensures the accept-and-commit is race-free. Producers stay concurrent either way.

// the same append use-case, unchanged across platforms:
const append = (req) => StreamMutex.serialize(
  Effect.gen(function* () {
    const meta = yield* StreamStore.readMeta            // read
    yield* validate(meta, req)                          // validate (content-type, seq, producer)
    yield* StreamStore.commit(buildMutation(meta, req)) // commit (atomic)
    yield* StreamHub.publish({ _tag: "Appended" })      // notify AFTER commit
  }),
)
// Cloudflare:  StreamMutex.serialize = semaphore.withPermits(1)  (fibers interleave at yield points)
// Node:        StreamMutex.serialize = semaphore.withPermits(1)
// Postgres:    StreamMutex.serialize = run-in-transaction

5. StreamRegistry — the per-stream context, and the one true seam

StreamService's effects require the substrate Services bound to one specific stream key. Providing that — for the right key, serialized, created on demand, evicted when idle — is the substrate's core job. StreamRegistry.withStream(key, effect) is the single function that supplies those Services to an effect, and it is the only thing that structurally differs between platforms:

flowchart TD
  H["core router: StreamRegistry.withStream(key, StreamService.append(req))"]
  subgraph cf["Cloudflare — the platform IS the registry"]
    do["one DurableStreamObject per stream (idFromName routes)"]
    dol["withStream = re-provide THIS instance's Services<br/>(StreamStore/Hub/Sched/ForkPeer already bound; StreamMutex = Semaphore(1))"]
    do --> dol
  end
  subgraph host["Node / Lambda — registry built in user space"]
    rc["StreamRegistry = RcMap&lt;key, context&gt;"]
    ctx["withStream = build/look-up the Services for key,<br/>idle-evict the context (flush + unsubscribe)"]
    rc --> ctx
  end
  H --> cf
  H --> host
Loading
  • Cloudflare: idFromName(path) already routed the request to the one instance that owns the stream, and that instance's ManagedRuntime already holds its Services. So withStream just re-provides the instance's context — the DO is the registry. Zero registry code.
  • Node / Lambda: one process serves many streams, so you build the actor-per-key yourself with Effect's RcMap (a reference-counted map of scoped values — one live context per key, released on idle: hibernation in user space). withStream looks up or builds the key's context and provides it. (Confirm RcMap in your effect version; otherwise a hand-rolled Map<key, Scope> + finalizers.)

Because the router always goes through StreamRegistry.withStream, StreamService and the router are byte-identical on every platform — the DO provides the per-stream Services statically, the registry provides them dynamically, and the core never learns which world it's in.

6. The capability matrix — same Services, different Layers

Service (Tag) Cloudflare Layer Node + FS (single host) Lambda (multi-instance) Memory (tests)
StreamStore @effect/sql-sqlite-do over co-located SQLite (lazy schema; delete = deleteAll) append-only file + meta.json; commit = journal→apply S3 (chunks + manifest) or DynamoDB Map
StreamMutex per-instance Semaphore(1) (fibers yield every 2048 ops) per-stream Semaphore(1) DB transaction / conditional write / advisory lock per-key Semaphore(1)
StreamHub in-proc PubSub in-proc PubSub + chokidar watch poll messagesAfter, or SNS→SQS / Redis in-proc PubSub
ExpiryScheduler storage.setAlarm setTimeout + persisted next-expiry EventBridge / DDB TTL / sweeper identity (lazy-only)
ForkPeer DO RPC getByName(src) local StreamStore(src) via registry local store + lock on src local store
StreamRegistry the DO itself (re-provide) RcMap per key RcMap per warm instance RcMap
substrate code minimal moderate most minimal

The interesting column is Lambda multi-instance: two of the DO's three donated properties now cost real work — a distributed StreamMutex (two warm Lambdas can target one key) and a non-local StreamHub (reader and writer may be on different instances). Honest options for the Hub:

  • Poll-backed (zero infra): the core's live=long-poll/sse polls StreamStore.messagesAfter every ~250 ms until data or timeout. Fully conformant, less efficient. Universal fallback.
  • External bus (real-time): Redis pub/sub, Postgres LISTEN/NOTIFY, SNS→SQS, Ably.
  • Pin streams to instances with a consistent-hash router in front — i.e. rebuild Durable Objects, which is the point.

Payoff: because read.ts/sse.ts target the StreamHub Service, a poll-backed Layer satisfies them with no change to the core — "live" degrades from push to fast-poll by swapping one Layer.

7. Composition & entrypoints (no # fields)

// core/layer.ts — substrate-agnostic; R = the six substrate Services
export const CoreLive: Layer.Layer<StreamService, never, SubstrateServices> =
  Layer.merge(StreamService.Default, StreamConfig.Default)

Cloudflare entrypoint — one runtime per stream; the DO is the registry

export const makeCloudflareLive = (ctx: DurableObjectState, env: Env) => {
  const adapter = Layer.merge(DurableStateLayer(ctx), EnvLayer(env))
  const sqlite = SqliteClient.layer({ storage: ctx.storage })   // @effect/sql-sqlite-do

  const perStream = Layer.mergeAll(
    StreamStoreSqlLive,          // Effect SQL client ← SqliteClient + DurableState(ctx)
    StreamMutexSemaphoreLive,    // serialize = Semaphore(1).withPermits(1)
    StreamHubPubSubLive,
    ExpirySchedulerAlarmLive,    // ← DurableState(ctx)
    ForkPeerRpcLive,             // ← Env(env)
  ).pipe(Layer.provide(adapter), Layer.provide(sqlite))

  const registry = StreamRegistryReprovideLive.pipe(Layer.provide(perStream))
  return {
    httpAppLayer: Layer.mergeAll(routes, registry, config),
    rpcLayer: Layer.merge(perStream, config),
  }
}

export class DurableStreamObject extends DurableObject<Env> {
  private readonly handler: (request: Request) => Promise<Response>
  private readonly rpc: ManagedRuntime.ManagedRuntime<PerStreamServices | StreamConfig, never>

  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env)
    // Schema is created lazily on first write; a drive-by read leaves storage
    // empty so the object ceases to exist on eviction. The shared memoMap means
    // the HTTP handler and the alarm/fork-RPC runtime build the per-stream
    // substrate ONCE (one PubSub / mutex / SQL client).
    const live = makeCloudflareLive(ctx, env)
    const memoMap = Layer.makeMemoMapUnsafe()
    this.handler = HttpRouter.toWebHandler(live.httpAppLayer, { memoMap }).handler
    this.rpc = ManagedRuntime.make(live.rpcLayer, { memoMap })
  }

  fetch(request: Request) { return this.handler(request) }
  alarm() { return this.rpc.runPromise(StreamService.runExpiry) }
}

Node entrypoint — one process runtime; the registry builds per-key Services

export const MemoryLive = Layer.mergeAll(
  StreamStoreMemoryLive, StreamMutexSemaphoreLive, StreamHubPubSubLive,
  ExpirySchedulerTimerLive, ForkPeerLocalLive, StreamRegistryRcMapLive,
)

NodeRuntime.runMain(
  HttpServer.serve(router).pipe(
    Layer.provide(NodeHttpServer.layer(createServer, { port: 8787 })),
    Layer.provide(Layer.mergeAll(CoreLive, MemoryLive, TelemetryLive)),
    Layer.launch,
  ),
)

The only differences: which substrate Layers get merged, and the StreamRegistry Layer (re-provide vs RcMap). CoreLive, router, StreamService, and every domain/* module are identical. Note that adapter-internal Services like Sql, DurableState, Env sit below the substrate Services (supplied via Layer.provide) and never appear in StreamService's R — they don't leak up.

8. The consistency contract, stated once and met per platform

The core requires this from StreamMutex + StreamStore:

  1. A mutation runs inside StreamMutex.serialize ⇒ no two mutations to one key interleave.
  2. StreamStore.commit(mutation) is all-or-nothing.
  3. Publish happens after commit returns ⇒ live readers never observe rolled-back data.
Platform (1) serialize critical section (2) atomic commit
Cloudflare DO per-instance semaphore (fiber yields break single-thread exclusivity) write coalescing across the I/O-free turn
Node + FS per-stream semaphore journal record (fsync) → apply → truncate; recover on boot
Postgres advisory lock / FOR UPDATE a real transaction
DynamoDB conditional version bump TransactWriteItems
Memory semaphore synchronous map swap

This is stronger than the DO-coupled original, which relied on DO's incidental coalescing. Naming the serialized region (StreamMutex) and the atomic unit (StreamStore.commit) lets the same core be correct on substrates with no implicit coalescing at all.

9. Forks across the substrate boundary

Fork touches two streams, so it stresses the model:

  • Cloudflare: streams are storage-isolated, so reaching the source needs RPC (ForkPeerRpc → getByName(src).exportFork). Cross-DO non-atomicity is handled as today: source ref-count++ before target commit; Effect.ensuring rolls it back on target failure.
  • Shared-store platforms (FS/Postgres/DDB): the source isn't isolated, so ForkPeerLocal opens StreamStore(src) through the registry and runs the same exportFork logic locally — but now it must take the source's StreamMutex (free on the DO). Same protocol code; only "how do I reach the source" differs.

10. Transport is already portable

@effect/platform's HttpRouter/HttpServerRequest/HttpServerResponse are substrate-neutral. Two seams cover every target:

  • Web-handler — HttpApp.toWebHandlerRuntime(rt)(router) → (Request) => Promise<Response>. Used by Workers, Deno, Bun, and Lambda Function URLs (shim the Lambda event ↔ Request/Response; awslambda.streamifyResponse for SSE).
  • Long-running server — @effect/platform-node NodeHttpServer.serve(router) (or BunHttpServer).

So SSE/long-poll bodies (HttpServerResponse.stream) work on Workers and Node alike; only the outer host adapter changes.

11. One core, many substrates, one oracle

The conformance suite (@humanlayer/durable-streams-server-conformance-tests, 232 black-box HTTP tests) targets an endpoint, so the same suite validates every platform Layer: run it against alchemy dev (Cloudflare), the Node+FS server, and the in-memory build. @effect/vitest covers the core once with MemoryLive + TestClock; a shared contract-test covers each substrate Layer ("commit is atomic under concurrent writers", "StreamHub notifies only after commit").

flowchart LR
  Suite["conformance suite (232 HTTP tests)"]
  Suite --> A["CloudflareLive @ alchemy dev"]
  Suite --> B["MemoryLive (Node)"]
  Suite --> C["NodeLive (FS)"]
  PT["substrate contract-tests"] --> A & B & C
Loading

12. Honest tradeoffs

  • Cloudflare is the cheapest precisely because it donates commit-serialization, co-located notification, and per-key lifecycle. Porting off DO means paying for those in infra or in degraded liveness (poll vs push). The design makes that cost visible and localized (three substrate Services), not hidden and pervasive.
  • Multi-instance Lambda is the hardest (distributed StreamMutex + non-local StreamHub, or consistent-hash routing). Single-host Node+FS is easy and behaves almost like a DO. Be clear which "Lambda" you mean.
  • StreamRegistry/RcMap is the code the DO saves you from writing (lifecycle, idle eviction, finalizer flushing). Generic and reusable, but it is an extra moving part off-DO.
  • StreamStore.commit atomic boundary reshapes today's fine-grained writes — worth it for portability and provability.

13. TL;DR

  • It's all Services and Layers. Core Services (StreamService, StreamConfig) have one canonical Layer; substrate Services (StreamStore, StreamHub, ExpiryScheduler, ForkPeer, StreamMutex, StreamRegistry) have one Layer per platform. A platform is a composite Layer (CloudflareLive / NodeLive / MemoryLive).
  • A Durable Object is a managed per-stream serialized actor: it donates request routing, co-located notification, and per-key lifecycle, so its Layers are trivial. (Commit serialization it does not donate under Effect — fibers interleave at scheduler yield points, so StreamMutex is a real Semaphore(1) on the DO too.)
  • StreamMutex is not a single-writer cap — the protocol stays multi-producer (Stream-Seq + producer fencing). It only serializes each append's read→validate→commit critical section; on any single host it's a semaphore, on a shared database a transaction / CAS.
  • StreamRegistry.withStream is the one true seam — identity-reprovide on Cloudflare, RcMap on a host — and it's why the router and StreamService never change.
  • @effect/platform makes transport portable, and one conformance suite validates every platform Layer.