Skip to content
C.W.K.
Lesson 06 of 06 · published

The Moment's Event Stream and the Engine's Shape

~14 min · server-sent-events, asyncio, launchd, deployment

Level 0First Word
0 XP0/47 lessons0/13 achievements
0/100 XP to next level100 XP to go0% complete

Fastest first needs a push

One of Spark's invariants is fastest first: show each layer the moment it exists. Pippa's reading of a menu streams in line by line, and the engine records each line as a result.add as soon as it parses. A phone that polled every few seconds would still see those lines arrive in clumps. So the engine pushes. Every op that names a moment goes straight to whoever is watching that moment.

Publishing from the one door

You have already seen where this starts: the last step of Corpus.record. Any op that carries a moment_id, or is itself a moment.* op, is published under that moment as {"type": op name, "record": body}. That rule leaves a gap: mark.remove and link.remove carry only their own id, so un-keeping a phrase or unlinking a crumb reaches no watcher until the client reads the moment again. Turn ops are special. They go out as the mirror holds them, status included, because some turns are started by the engine on its own (a cold review that follows a finished version) and a client learns about those only through this event.

A few events are deliberately not corpus. Corpus.publish sends a transient event to the watchers and records nothing. The practical turn uses it for turn.prose, Pippa's words to Dad as they grow around the structured results, and the show-card interpreter uses it for its exchange.* progress events. A client that misses one misses progress, and the practical turn still records its finished prose when it ends.

One process, many loops

The fan-out is a small in-process class, Broadcaster, in the first code block. It maps a moment id to a set of subscribers, and each subscriber is a pair: the event loop that created it and an asyncio.Queue.

The pair is the interesting part. record runs on a worker thread, because corpus writes are pushed off the event loop through asyncio.to_thread. An asyncio queue is not thread-safe, so the worker thread must not put into it directly. publish copies the subscriber list under a plain threading lock, then hands each delivery to the subscriber's own loop with call_soon_threadsafe. If that loop has already closed, the call raises RuntimeError and the subscriber is dropped quietly.

No broker, no Redis, no message bus. That is enough because one engine process is the only writer; every event that can exist is born inside it.

The stream itself

The second code block is the whole route, GET /api/moments/{id}/events, served as Server-Sent Events:

  1. An unknown moment answers 404 before any stream opens.
  2. It subscribes first, then reads a snapshot of the moment and sends it as the first event. In the other order, an op recorded between the read and the subscribe would be in neither. In this order it may appear in both, and that is harmless because clients key results by id.
  3. Then it forwards events as they arrive. After 15 seconds of silence it sends : keepalive, an SSE comment that clients ignore and that keeps the connection from looking idle.
  4. When the client goes away the generator is closed, and finally unsubscribes.

Two response headers matter: Cache-Control: no-cache, and X-Accel-Buffering: no so a proxy in between does not hold events back in a buffer. A test subscribes to a moment, records a result from a worker thread, publishes a turn.prose, and asserts both arrive in order, then checks that nobody is still watching once it unsubscribes.

The browser's EventSource reconnects by itself; the phone opens the same route from Swift. Since the phone reaches the engine only over the tailnet, a lost moment stream on the phone tells Dad to check Tailscale, and the web adds the same hint to a failed fetch.

The engine's shape

ConcernHow Spark does it
ProcessOne launchd LaunchAgent, com.cwk.spark-serve, running python -m backend.serve with RunAtLoad, KeepAlive and a 10-second throttle; output goes to serve.log under the data root
Listeninguvicorn on 127.0.0.1, on the port the family registry gives Spark (7900)
ReachTailscale Serve publishes an HTTPS twin (7901) onto that loopback port, plus an HTTP twin where the registry declares one
SiblingscwkPippa, the family control facade and cwkBellows, each dialled on loopback at the origin the registry gives
Web clientThe built frontend mounted last, so every /api route wins over the page fallback
SchedulesNone. The data root is covered by the family's daily backup pass

The settings module explains the loopback rule in its docstring: nothing there may name a LAN host, because a launchd agent cannot pass macOS Local Network gating. The engine never listens on a wildcard address; the tailnet reaches it only through Serve. Every port, origin and data root comes from the family registry, and environment variables remain as explicit overrides. Even the tracked plist is a template, with tokens such as @@HOME@@ and @@DATA_ROOT@@ that the install script renders for the Mac it runs on.

An install that checks where it is

The install script is generated from the family kit's template, one shape for every engine (Sibling Kit Quest teaches why vendored copies beat a shared package). The third code block is its first act: ask the registry whether Spark belongs on this Mac, and refuse to continue if not. Then it builds the conda environment and the frontend, renders the plist, boots the old job out and waits past its exit timeout, bootstraps the new one with retries, and polls /api/health fifteen times, two seconds apart. Last, it publishes the tailnet twins through a second script that asks the registry the same question before it touches Tailscale.

One line in it records a lesson the family paid for: there is no launchctl kickstart -k after the bootstrap. A fresh load of a RunAtLoad job has already started it, and the kickstart that used to follow killed that first boot, possibly in the middle of a database migration.

Principle: Let the machine that runs a service be a fact someone else owns. Spark reads its port, its data root, its neighbours and even its permission to install from the registry, so moving it is a registry decision, never a code change.

Code

cwkSpark, backend/events.py: deliver on each subscriber's own loop (verbatim, lines 45-52)·python
    def publish(self, key: str, event: dict[str, Any]) -> None:
        with self._lock:
            watchers = list(self._subscribers.get(key, ()))
        for loop, queue in watchers:
            try:
                loop.call_soon_threadsafe(queue.put_nowait, event)
            except RuntimeError:
                self.unsubscribe(key, (loop, queue))
cwkSpark, backend/app.py: the moment as it is, then every change (verbatim, lines 721-751)·python
    @app.get("/api/moments/{moment_id}/events")
    async def moment_events(moment_id: str) -> StreamingResponse:
        """The moment as it is, then every change as it happens (SSE)."""
        if await asyncio.to_thread(corpus.moment, moment_id, include_deleted=True) is None:
            raise HTTPException(status_code=404, detail="moment not found")

        async def stream() -> AsyncIterator[bytes]:
            # Subscribe before the snapshot so nothing falls between them;
            # clients key results by id, so a repeat is harmless.
            subscriber = corpus.events.subscribe(moment_id)
            try:
                snapshot = await asyncio.to_thread(
                    corpus.moment, moment_id, include_deleted=True
                )
                yield _sse({"type": "snapshot", "moment": snapshot})
                queue = subscriber[1]
                while True:
                    try:
                        event = await asyncio.wait_for(queue.get(), timeout=15)
                    except asyncio.TimeoutError:
                        yield b": keepalive\n\n"
                        continue
                    yield _sse(event)
            finally:
                corpus.events.unsubscribe(moment_id, subscriber)

        return StreamingResponse(
            stream(),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
        )
cwkSpark, the service install script in scripts/: refuse unless the registry places Spark here (verbatim, lines 13-21)·shell
# This script lives in the repository's scripts/; the repository is its member
# id in the family registry, which owns the engine's port and tailnet URLs.
REPO="${0:A:h:h}"
# Only on the Mac the family registry places this member on, including a
# machine's host profile override (cwkPippa todo #549, Pippa's Ark).
/usr/bin/python3 "$REPO/scripts/kit_family_endpoints.py" runs-here "${REPO:t}" || {
  echo "${0:t} runs only where the family registry places ${REPO:t}" >&2
  exit 1
}

External links

Exercise

Add a live view to a service of yours. Write a per-key broadcaster whose publish can be called from any thread and delivers through call_soon_threadsafe onto each subscriber's loop. Expose it as a Server-Sent Events route that subscribes before it reads a snapshot, sends the snapshot first, and sends a comment line after 15 seconds of silence. Write a test that publishes from a worker thread and asserts arrival order, and another that inserts a write between subscribe and snapshot and shows the client ends with exactly one copy after merging by id. Finally, give the service's install script a first step that refuses to run on a machine your configuration does not assign it to.
Hint
Store each subscriber as the pair (loop, queue) captured when it subscribes. Unsubscribe in a finally block so a closed connection always cleans up, and let a RuntimeError from a closed loop drop the subscriber.

Progress

Progress is local-only — sign in to sync across devices.
← PreviousTurns Survive a RestartNext →Quiz · 4 questions
Spotted a bug or have feedback on this page?Report an Issue

Comments 0

🔔 Reply notifications (sign in)
Sign in — Please sign in to comment.

No comments yet — be the first.