# 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/contoso-fixtures/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`](../internal/store/onelake.go) already emits a `FileEvent` after every committed write, rename and delete — added for event triggers ([`internal/api/triggers.go`](../internal/api/triggers.go)), but the choke point is general: ```go 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 ```go // 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. ```json { "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: `) 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`](../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](#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//_delta_log/…0004.json` **is** a table-version event. The commit's own `add`/`remove` actions (already parsed by [`internal/warehouse/delta.go`](../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`](../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`](../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`](../internal/api/pipelines.go)), and already records both on the lineage edge. So: ```go 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=` 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](demo/README.md#regenerate-flowgif) 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`](../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: ```bash 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](27-running-modes.md#the-terminal-profile-needs-two-things) 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 `` 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 `