Flow observability: watching data move through the emulator

Status: built. All four steps — the event bus, every event kind, the SSE endpoint, attribution, and the portal Data flow view.

Running the medallion example today is a black box. It prints step numbers, and when something fails you reconstruct what happened afterwards from queryactivityruns — which is why examples/medallion-pyspark/common.py grew hand-rolled activity-level failure reporting. You cannot watch the data move, and you cannot see a failure at the moment it happens.

That is a tooling gap, not a missing capability. Every fact needed is already recorded; none of it is streamed, and the pieces are not joined.

The one thing that makes this cheap

The emulator owns its storage layer, so every byte that moves through OneLake passes through one function. internal/store/onelake.go already emits a FileEvent after every committed write, rename and delete — added for event triggers (internal/api/triggers.go), but the choke point is general:

type FileEvent struct {
  Type        string // Microsoft.Fabric.OneLake.File{Created,Deleted,Renamed}
  WorkspaceID string
  ItemID      string
  RelPath     string
}

No writer can bypass it — not an ADLS client, azcopy, delta-rs, Sail, a Copy activity, or the mirror writer. A real Fabric would need an Eventstream and a broker to get this; we get it from owning the storage layer.

What was missing is only that Store.FileEvents was a single func field with one subscriber. Everything below follows from making it a fan-out.

Two contracts, deliberately kept separate: FileEvents stays synchronous and exactly-once (a Reflex must have fired before the write returns), while the bus is asynchronous and lossy (a watching developer must never be able to slow a writer down).

Design

1. An event bus in the store

// Subscribe returns a channel of events and a cancel func.
func (s *Store) Subscribe() (<-chan Event, func())

A channel rather than a callback, in the end: the SSE handler already selects over the request context and a keepalive ticker, and a callback would have needed a channel behind it anyway.

FileEvents becomes the bus’s first publisher rather than the only consumer.

The critical constraint. The emit is synchronous inside the write path, and the store runs SetMaxOpenConns(1). A subscriber that blocks stalls OneLake writes for every caller. So:

• each subscriber gets a buffered channel (256) and its own goroutine;
• a full buffer drops the event and increments a counter, it never blocks;
• the consumer collects that count (Subscription.TakeDropped) and reports it, so a gap is visible rather than silent.

Why the consumer reports it, and not the bus. The first cut had the bus announce drops by injecting a dropped event during dispatch. That is subtly broken: dispatch only runs when an event arrives, so a subscriber that falls behind and then goes quiet is never told — exactly when it most needs to know. CI found it as a flaky test on one platform; the test was racy because the guarantee was. The count is now available the instant a drop happens, and the SSE handler polls it on every event and every keepalive, so a gap surfaces within one interval whether or not traffic continues.

Slow consumers degrade themselves, never the emulator. This is the one place a mistake here would be expensive, so it is stated first.

2. Event kinds

One envelope, seven kinds. seq is a monotonic per-process counter so a client can detect gaps; at is emulator time (the controllable clock), because every other timestamp in the system is.

{ "seq": 1041, "at": 1754049600, "kind": "…", "…": … }

kind	emitted when	carries
file	a OneLake path is written / renamed / deleted	eventType, workspaceId, itemId, path, attribution
table	a Delta commit lands (see below)	itemId, table, version, rowsAdded, filesAdded, filesRemoved, attribution
activity	a pipeline activity starts or finishes	jobId, activityName, activityType, status, error, durationInSeconds, retryAttempt
job	a job instance starts or reaches a terminal state	workspaceId, itemId, jobId, jobType, invokeType, status, failureReason
lineage	a source→target movement is recorded	workspaceId, itemId, table (the target path), producer
query	a semantic model is read	workspaceId, itemId, table (the model), activityType (the caller)
dropped	a subscriber fell behind	dropped (how many it missed) — see the ring buffer below

dropped is the odd one out and deliberately so: it reports on the subscriber, not on the platform. store.ViewKinds is the other six, and a UI offers those as filters — switching off the one signal that says the log is incomplete is not a filter worth having.

The client’s list is generated, not written. The stream names every frame (event: <kind>) and EventSource has no wildcard listener, so a kind missing from a client’s subscription list is invisible: no error, no dropped count, nothing. scripts/gen_event_kinds.py therefore generates portal/src/eventKinds.ts from store.AllKinds, store.ViewKinds and the store.Event struct — the kinds, their one-line descriptions, and the wire shape as an interface. make check fails when the committed file and the Go disagree, and because portal/dist is embedded in the binary the two always ship together, which makes drift impossible rather than merely detected.

Because it is generated as TypeScript, the guarantee reaches past subscribing: EventKind is a union, Flow.svelte’s describe() switches over it exhaustively, and ingest() has a never arm for a kind that is neither dropped nor renderable. Declare a kind in Go, regenerate, and the portal stops compiling until something renders it. make portal-types proves that claim by declaring a kind nothing handles and requiring the build to fail — a default: arm added in good faith would otherwise restore the silent-loss bug with every other check still green.

Existing shapes are reused verbatim where they exist — activity is pipeline.ActivityRun plus the job id, job is the jobBody fields. Nothing new to learn, and no second source of truth for a status.

The last two exist because a medallion’s final hops have no job to end and no OneLake write to observe — see The hops that are not OneLake writes. A query is deliberately an event and never an edge: reading a model moves no data.

When an activity is announced. Not when it is recorded. The interpreter’s retry loop discards failed attempts and back-patches the survivor’s retryAttempt, durationInSeconds and — on a timeout — its status, so announcing at record time would stream outcomes that never appear in the run, and stream them before they were final. Instead the interpreter flushes at points where its records are settled: after runWithPolicy returns, and after the skip / unresolvable-dependency records that bypass it. The stream and queryactivityruns therefore always agree, which is the design’s rule.

When a job is announced. Started always. A terminal event only where one genuinely exists: a DataPipeline (which runs inline), a Dataflow (which fails loudly), a notebook / Spark job / Airflow DAG at the moment its engine reports back. A generic item’s status is derived from the clock and never has such a moment, so nothing further is claimed for it. The stream says what happened and stays quiet where nothing did.

3. Delta commits are what make the stream legible

Raw file events are a firehose: one table write is dozens of Parquet parts plus a log entry. Unfiltered, that is noise, not a visualization.

But a write to Tables/<name>/_delta_log/…0004.json is a table-version event. The commit’s own add/remove actions (already parsed by internal/warehouse/delta.go) say what changed. So the bus watches for _delta_log writes and derives a table event:

bronze_customers → v4  (+1,203 rows, 1 file added, 0 removed)

TableRoot() in internal/onelake/observe.go already collapses part-file paths to the table root, so the grouping exists.

A consumer can therefore watch table events alone and see the medallion move, or drop to file events when debugging what a writer actually did. Both are published; the client chooses. A ?kinds=table,activity,job filter keeps the common case quiet.

4. Attribution: which activity moved these bytes

A FileEvent says what moved, not who moved it. Two mechanisms already answer that, and this design unifies them rather than inventing a third:

• Notebook / Spark writes — x-ms-fabric-job-id / x-ms-fabric-cell-index headers, or the same values as bearer extraClaims for engines built on Rust object_store (delta-rs, Sail) which cannot set request headers. observe() in internal/onelake/observe.go already computes this and throws it away after recording the access.
• Copy activity — the executor knows its own jobID and the activity’s Name at the moment it writes (internal/api/pipelines.go), and already records both on the lineage edge.

So:

type Attribution struct {
  JobID        string
  ActivityName string // Copy and other pipeline activities
  CellIndex    *int   // notebook cells
}

CellIndex ended up a pointer, not an int with a sentinel: cell 0 is a real cell, and a plain int cannot tell “the first cell” from “not a cell at all”. Same reason Version is a pointer on a table event.

How it reaches the write, without plumbing context through everything. CreateOneLakePath is called from the DFS and Blob handlers, git, deployment, the mirror, and the Delta writer. Threading a context through all of them to serve two callers would be a large diff for a small gain. Instead:

• CreateOneLakePath keeps its signature and emits an unattributed event;
• a sibling CreateOneLakePathAs(attr Attribution, …) emits an attributed one;
• warehouse.WriteDeltaTableAs is the same pattern one level up, so the Delta writer’s two writes — the Parquet part and the _delta_log commit — both carry it, and the derived table event inherits it from the commit;
• the OneLake handlers pass attributionOf(r), which is the value observe() was already computing and discarding.

Sibling functions rather than a parameter, so the ten existing callers and their tests stay untouched; only the sites that know something change.

Explicit, no globals, no goroutine-local trickery, and every other caller is untouched. Attribution is never inferred — same rule the lineage design already holds to: an engine reports it, or the data plane observes it, or the field is empty.

Note what this does not change: lineage_edges remains the authoritative record of a source→target movement. Attribution on a file event is a live debugging aid; the lineage edge is the durable fact.

4a. When a file event fires — a bug this work surfaced

Building the stream exposed a real defect in the already-shipped event triggers. The ADLS protocol writes in three steps — create the path, append the bytes, flush — and the event was firing on the create, while the file was still empty. A Reflex therefore triggered before its data existed, and a derived table event would have reported an empty commit.

Azure’s own ADLS Gen2 raises BlobCreated on FlushWithClose, not on the create, and the DFS handler’s own comment already said flush “is the point the DFS protocol considers the file written”. So:

• a zero-length non-directory create raises nothing — mid-sequence it is indistinguishable from an empty file, and the store cannot tell them apart;
• the flush handler calls Store.EmitFileWritten, because only the protocol layer knows a staged write is complete.

Exactly one event per file, carrying data that is actually there.

5. The stream: GET /_emulator/events

Server-Sent Events, not a WebSocket:

• one-way is all this needs;
• curl -N tails it with no client at all — which is the single biggest quality-of-life win here, and it needs no portal work;
• no new dependency, ~40 lines of Go;
• browsers reconnect automatically via EventSource.

curl -N https://localhost:9443/_emulator/events?kinds=table,activity,job

Real output, captured from the test suite — an ADLS upload followed by a Delta commit:

event: file
data: {"seq":1,"at":1785591720,"kind":"file","workspaceId":"3cfca386…","itemId":"bf775fce…",
       "eventType":"Microsoft.Fabric.OneLake.FileCreated","path":"Files/landing/customers.csv"}

event: file
data: {"seq":2,…,"path":"Tables/bronze_customers/_delta_log/00000000000000000000.json"}

event: table
data: {"seq":3,…,"table":"Tables/bronze_customers","version":0,"rowsAdded":1203,"filesAdded":1}

version is deliberately a pointer in Go so a table’s first commit — version 0, the common case for a fresh medallion — is still reported, while file events carry no meaningless "version": 0.

It sits under /_emulator, the existing testing-lever namespace (clock, faults, portal) — deliberately not part of the Fabric contract, because real Fabric has no such endpoint. It is emulator-only, and docs/parity.md records it in the emulator-only table alongside the clock and the fault injector.

Ring buffer. The bus keeps the last 1,000 events, and ?since=<seq> replays from there. A client that connects after the run started still sees it, and the portal survives a reload mid-medallion.

6. The portal Flow view

The README’s hero image is this view, generated rather than screen-recorded: docs/demo/flow.py films the empty graph, then seeds a medallion behind it. Reproducible, so a layout change that spoils the framing shows up before it ships.

A new entry under Data plane in portal/src/App.svelte, which already has the section structure:

• A live log — the event stream, filterable by kind and by workspace. Failures in red, with the activity error inline. This alone replaces reading container logs.
• A flow graph — nodes are items and tables, edges come from lineage_edges (which already carry producer, so a Copy edge and a notebook edge are visually distinct). A node written in the last ten seconds is bright, one written earlier in the session keeps a quieter mark, and one whose writing activity failed goes red. Two levels rather than one because they answer different questions — what just changed and what this run has touched at all; a single state means a finished run is uniformly lit and says nothing. On the medallion this draws itself: source → landing → bronze → silver → gold → semantic model, with a Power BI read pulsing at the far end.
• A table inspector — select a node for the current Delta version, schema, row count and a 20-row sample, read through the same warehouse reader the SQL endpoint uses (/_emulator/portal/table). The stream says a table changed; this says what it changed into, which is the question a developer asks next. Re-read automatically when a table event lands on the open table, so the panel cannot quietly go stale. A Files/ node says plainly that there is no schema to read rather than erroring, and a table whose first commit has not landed reports that as a fact, not a 500.

Polling would have worked for the log. It would not have worked for the graph: the point is watching it happen.

Nodes are laid out in columns by distance from a source — computed by relaxation rather than a topological sort, so a cyclic graph still renders instead of hanging — which makes a medallion draw itself: landing → bronze → silver → gold → semantic model.

An edge is drawn straight from its producer, and the styling encodes how well the movement is known rather than merely who caused it:

Producer	Drawn	Because
Copy	solid	the emulator’s own executor moved the bytes
Warehouse, NotebookObserved	solid, in the success colour	the emulator watched it happen — evidence, not a claim
Notebook, Reported	dashed	an engine or a step reported the movement
DirectLake	finely dotted	a binding, not a copy: the model reads the Delta where it lies, so the hop is real but no bytes move

A reader can therefore tell at a glance which parts of a graph are observed and which are asserted, which is the distinction the whole design rests on.

The hops that are not OneLake writes

Three parts of a medallion move data without writing a byte through OneLake, and each needed its own answer rather than an inference:

Hop	How it is known	producer
silver → gold	the TDS front parses each statement it forwards and records what the engine ACCEPTED (internal/tsql.DataFlows, internal/server/warehouselineage.go)	Warehouse
gold → semantic model	a Direct Lake table’s binding names its source, so the edge is recorded when the definition lands	DirectLake
a source system → landing	the ingesting step names the CONNECTION it authenticated through (connectionId on a read). A medallion does not begin in Fabric, and until this existed the first node the graph could draw was a file already sitting in Files/landing — the vendor that put it there was unsayable. The connection is used rather than a URI because it already exists, carries a display name, and is what the client actually authenticated through; the emulator resolves it and refuses an id that names nothing	Reported
bronze → silver, and an import model’s sources	the step reports its own derivations to POST /v1/workspaces/{wid}/lineage — a list of moves, each a real (reads → writes) group, because pairing one flat read list against one flat write list is a cross product that overstates: silver reads two bronze tables and writes three silver ones, but the quarantine comes from the orders alone	Reported

The producer is the point. Warehouse is evidence — the emulator watched the engine accept the statement — while Reported is a claim by the caller, and a consumer must never have to guess which it is holding. An IMPORT semantic model gets no automatic edge at all: its rows arrive in the definition already detached from wherever they were selected, and manufacturing a source would be the guess this design refuses everywhere else.

A query is not a movement, so Power BI consumption is a query event on the bus and never an edge. lineage_edges stays a record of things that moved.

Because a warehouse build has no job, the graph reloads on a lineage event rather than waiting for a job to reach a terminal state — coalesced, since one dbt model emits an edge per source and they all describe the same redraw.

The graph reads /_emulator/portal/lineage, a tenant-wide listing added for this view: the API-facing /v1/workspaces/{id}/lineage is workspace-scoped because it sits behind RBAC, and the portal has no principal.

The terminal pane

Driving the pipeline while watching it run, in one window: the Flow view can open a shell beside the graph. Off unless you ask for it, and it needs two opt-ins — the terminal profile to start ttyd, and docker-compose.terminal.yml to tell the emulator where it is:

docker compose --profile terminal \
  -f docker-compose.yml -f docker-compose.override.yml -f docker-compose.terminal.yml \
  up -d

The emulator prints a token once at startup; paste it into the pane. It is not served by any endpoint, because the portal is unauthenticated and an endpoint handing the token out would be the same as having none.

Why the route carries its own auth at all. The other portal routes are reads over local state. A terminal is arbitrary execution, so it does not inherit that premise: internal/server/terminal.go demands a bearer, and the status endpoint dials ttyd rather than trusting configuration — a stack whose profile is off says so instead of offering a pane that dies when clicked.

See 27-running-modes.md for the whole-stack picture, including what naming -f costs you.

What this deliberately does not do

It does not pack every emulator into one UI. entra, Key Vault, Sail, SQL Server, Airflow, MLflow, Kusto and OpenMetadata are separate containers; several have their own UI and rebuilding them is a large surface that rots on every sidecar version bump. Health badges and outbound links, not embedded frames.

The fabric-emulator portal is the right hub for a different reason: it is the one component that sees everything — every OneLake byte, every job, every validated token. Sail’s writes already appear here, attributed, because of the bearer-claims path. That is a genuine cross-emulator view that costs no proxying.

It does not add a background worker. The bus is passive: it publishes what callers already do. Nothing polls, nothing ticks. The emulator’s determinism guarantee is untouched.

It does not become a second source of truth. Every event is a projection of state already persisted (job_instances, pipeline_runs, lineage_edges, onelake_paths). If the stream and the API ever disagree, the API is right.

Build order

1. Bus + file/table events + SSE endpoint. Done. Useful on its own: curl -N while the medallion runs. Also emitted Delta stats (numRecords) from our own writer, which every real Delta writer already does and without which our own commits could not report row counts.
2. activity/job events. Done. Failures arrive as they happen instead of being reconstructed afterwards — with the activity name, its error, and the job id that correlates them.
3. Attribution (§4). Done. A Copy activity’s writes name the activity; a notebook cell’s writes name the cell — including cell 0, and including engines that cannot set headers, via the bearer claims that were already there.
4. Portal Flow view. Done.

Each step is independently useful, and each stops at a point where the tree is green.

Testing

The bus is pure Go with no clock dependency, so it unit-tests directly: subscribe, write a file through the store, assert the event. The drop-on-full path is the one that matters most and is the easiest to get wrong — a deliberately blocked subscriber must not stall a write, and must report a non-zero drop count.

The SSE endpoint gets a server-level test that runs a real pipeline and asserts the expected event sequence arrives on the wire, in order, with seq gaps absent. The medallion e2e gains an assertion that the run produced table events for bronze_customers and the gold table — which also makes the example a witness that the stream reflects real work.

Per the repo rule, the docs/parity.md row and its docs/witnesses.json entry land in the same commit as the claim.

Portal styling

The portal now uses shadcn-svelte components on Tailwind 4, themed to Fluent rather than shipping shadcn’s default slate.

That last part is the point. The portal is the operator console for an emulator of a Microsoft product, and a developer should recognise where they are: Azure blue actions, 4px corners, white surfaces on a warm-grey canvas, the Segoe stack — the intent the original hand-rolled CSS recorded as “Fluent-mimic tokens per upstream DESIGN.md”. shadcn is used for what it is good at (accessible primitives, real focus rings, dark mode, a coherent token contract); the palette stays Fabric’s.

Every token a shadcn component reads resolves through @theme inline to the variables in portal/src/app.css, so a retheme is one block rather than a sweep through markup.

Light, dark, or the OS. The dark palette hangs off :root[data-theme='dark'], stamped on <html> by portal/src/theme.ts — and by an inline script in index.html before the bundle loads, so the first paint is already correct rather than flashing light. The toggle in the top bar cycles system → light → dark → system: three states, because “follow the machine” is a real answer that a two-way switch cannot express. It used to be a prefers-color-scheme media query with no switcher, on the reasoning that a local tool has nowhere to store a preference; the sidebar’s fold already disproved that, and a recording or a screenshot usually wants a chosen palette rather than whatever the author’s laptop was set to.

Every view is built from the components, not from a shared class vocabulary. The eleven views used to speak .card, .chip, .panel and a bare <button>, reimplemented on shadcn’s tokens. They now use the components themselves — Table, Card, Button, Input, Label, Checkbox, Textarea, Badge — so the accessibility and focus behaviour come from the primitives rather than from CSS that resembles them. Status colour goes through src/lib/StatusBadge.svelte, a wrapper of ours rather than a fork of the vendored Badge: shadcn-svelte add regenerates that file, so a success variant added to it would be overwritten by the next update. It carries data-tone, which is what a test asserts instead of reaching into Tailwind classes.

One deliberate exception: the workspace filter stays a native <select>. bits-ui’s Select is a popover-driven listbox that needs pointer capture and layout jsdom does not implement, and the trade — a better-looking control that the suite cannot drive — is not worth it for one filter. It is styled to match the other inputs.

Cost, stated: the bundle grew from 66 kB JS / 5.8 kB CSS to ~145 kB / 59 kB (43 kB / 10 kB gzipped) — bits-ui and its primitives. For a portal embedded in a local binary that is a fair trade for accessibility and consistency; it would not be on a public site’s critical path. Tailwind is now a build dependency of portal/, installed through the existing pnpm workspace, and CI’s pnpm install --frozen-lockfile covers it.