fix(realtime): NATS is not a boot dependency of a role that only publishes; a sync node without the bus fails in 15 s, coded; a publish with no live connection is refused, never queued

CHANGELOG.md

@@ -10,6 +10,18 @@ Semver applies from 1.0.0. A breaking change to a documented API needs a major 
 
 ### Fixed
 
+Tier 0 — core. Tier 1 — cache. Tier 3 — realtime. Tier 5 — cli.
+
+- **NATS is no longer a boot dependency of a role that only publishes to it.** Under `realtime: { transport: 'nats' }` every role awaited the first dial before it bound a socket, and that dial never gave up: `openNatsClient` set `waitOnFirstConnect`, under which nats@2.29.3 retries a failed first dial past its attempt counter, forever. A `web`, `worker` or `scheduler` pod that restarted while NATS was down (an OOM, a node drain, a rollout, a scale-up) therefore hung in boot: pages, webhooks and jobs were down for a bus those roles only send "re-read" events to, and `health.readiness: 'process'` could not help because no listener existed yet. The role now decides (`runtime-bus.ts`, `selectTransport(env, realtime, { use })`): `sync` and the replicator still need the bus and await it; `web`, `worker`, `scheduler` and the dev MCP host dial it in the background on a jittered backoff (500 ms doubling to 30 s) and boot at once. Nothing to change in an app. The table is [Realtime → What each role needs from the bus](https://github.com/developerz-ai/ultimate/wiki/Realtime#what-each-role-needs-from-the-bus).
+- **A `sync` node or a replicator without the bus fails fast and says why.** The dial is retried for 15 s (`BUS_CONNECT_WAIT_MS`) and then refused with `X_TRANSPORT_UNAVAILABLE`, whose cause names the server, the wait and the last attempt's own failure, and whose fix names the variable `realtime.urlEnv` points at. It used to sit silent until something killed it. Each attempt carries a 4 s connect timeout (`DIAL_TIMEOUT_MS`; the library's is 20 s), so a server that accepts the socket and says nothing still fails attempts inside the wait and the refusal's `last attempt:` is never empty.
+- **A publish with no live connection is refused at once, not buffered.** While the background dial is in flight, `publishChannelEvent` and the cache-invalidation broadcast reject with `X_TRANSPORT_UNAVAILABLE` immediately. A publish made after the client's socket has closed is refused the same way, in every role: the `nats` client buffered it without bound and replayed the lot when the server returned, and it has no option that forbids the buffer, so the transport reads the client's own connection flag before it hands a publish over. One window is left and is stated rather than hidden: a connection whose packets are silently dropped reads as connected until two pings go unanswered — now 10 s apart (`PING_INTERVAL_MS`, `MAX_PINGS_OUT`), so within 30 s where the library's default is six minutes — and a publish made inside it is buffered and sent on reconnect. Publishing resumes with no restart when the bus is back, on first connect and after every loss.
+- **A cache bust made while the bus is away reaches the other replicas when it returns.** The replica that made it, Redis and the CDN are cleared as before and the broadcast lands in the report's `errors` (`cache.invalidate.partial`). The refused wire tags are now kept — de-duplicated, at most 1,024 (`MAX_DEFERRED_TAGS`) — and published on the transport's next reconnect, or after the next publish the bus accepts; past the bound the process owes one flush-all marker (`{"flush":"all"}` on `x.cache.invalidate`) instead of a queue, so nothing grows with an outage. What is owed is published by one settle at a time and taken before its publish — put back if the bus refuses — so overlapping reconnects send a batch once and a bust refused mid-publish is never cleared by a success that did not carry it. And every process drops what it holds in its own heap when its connection comes back, or when its boot subscribe lands after a refusal: it heard no peer for a window nobody measured. That is `flushProcessTiers()` on `@ultimat3/cache` — `CacheTier.clear()` on the in-process tier (`lruTier` implements it; a shared tier does not), every tag-revalidated ISR page marked stale (a peer's bust reaches those pages by this broadcast and by nothing else), and every fill in flight fenced out. The cost is one cold in-process cache per replica per NATS outage. A replica running an older build ignores the marker and waits for its TTLs, as it always did.
+- **The replicator rides out a NATS restart.** A publish refused because the bus is away ended the run on the spot — stream stopped, lock released, backoff, re-read from the last lsn, `/readyz` false throughout — which the refusal above would have made the cost of every blip. The handler now retries the same envelope three more times over 2 s (`PUBLISH_RETRY_DELAYS_MS`), fenced after each wait, and only then ends the run for the takeover loop. `stop()` ends a wait in flight at once. A refusal that is not the bus being away (a payload over the server's limit) is not retried.
+- **Only a `sync` node requires JetStream at boot.** The presence bucket was asserted on every dial of every role, so a publisher failed on a nats-server without `-js`. It is asserted at the dial where presence is served and on first use of `transport.shared` everywhere else (`NatsTransportOptions.presenceBucket`).
+- **A dead bus no longer takes a publishing role out of the ingress.** `/readyz` in the default `'dependencies'` mode answered 503 on every `web` pod at once while NATS was down. A readiness check now has a third answer, `'degraded'` (`registerReadinessCheck(name, check, { onFailure: 'degraded' })`, `ReadinessStatus`): reported by name, never a 503 on `/readyz` in either mode, and a 503 on `/readyz?deep=1`. The `transport` check is `degraded` on `web`, `worker` and `scheduler` and stays `failing` on `sync` and the replicator. A monitor that must see a bus outage on a web pod reads `?deep=1`, as [Deployment](https://github.com/developerz-ai/ultimate/wiki/Deployment) already says to.
+- **The boot line says where the bus is.** `ultimate started` carries a `bus` field — `nats(up)`, `nats(connecting)` or `in-process` (`ServedApp.bus`) — and `ultimate bus` is logged with `bus: 'nats(up)'` each time the connection lands. A dial that keeps failing is reported on attempts 1, 2, 4, 8, … rather than once per attempt, and so is `cache.broadcast.subscribe-failed`; that subscribe is woken by the bus coming up instead of waiting out its own backoff.
+- **`realtime.enabled: false` never waits for the bus.** With no realtime role to start, no process of such an app awaits the dial, `ROLE=sync` included (it is refused by `rolesUnderRealtime`, as before, and now without first waiting on NATS). The connection is still opened in the background under `transport: 'nats'`, because cache invalidation between replicas crosses it.
+- **Types that changed, though nothing here is `BREAKING`.** Reading code is unaffected; code that *builds* or *exhausts* one of these needs an edit. `ReadinessStatus` is now `'ok' | 'degraded' | 'failing'`, so an exhaustive `switch` over it (a `never` default) stops compiling until it handles `'degraded'`. `TransportSelection` gained the required `use` and `state()`, so a hand-built one (a test double) must add both; `selectTransport` is the only thing in the framework that builds one. `ServedApp.bus` is optional. `CacheTier` gained the optional `clear()`. New names: `ReadinessCheckOptions`, `BusUse`, `BUS_CONNECT_WAIT_MS`, `NatsTransport.connectInBackground()`, `connect({ withinMs })`, `NatsTransportOptions.presenceBucket` / `dialTimeoutMs`, `flushProcessTiers`, `FLUSH_ALL_TAG`.
 Tier 4 — render.
 
 - **A locale-prefixed `isr` page revalidates.** The ISR store key keeps the routed locale prefix (`/en/blog/a?__x_locale=en`, one document per URL), and the lookup that finds the key's ROUTE stripped the query and not the prefix — so `/en/blog/a` matched no route while `/blog/a` matched `/blog/:slug`. Every prefixed page was stored with no `ttl` and joined no tag: `invalidateTags` and the route's own `revalidate.ttl` both expired the default locale's copy and left the prefixed one, which answered a withdrawn post with a 200 (or a 404 stored before publication) for the life of the process, advertised the tag-only `s-maxage=60` under a route that declared its own, and carried no `Surrogate-Key`, so a CDN purge missed it too. The route is now matched on the path with its routed prefix removed (`unlocalizedPath`, the split the router's own match is built on); the key is unchanged, so entries stay one per locale and a regeneration renders in the entry's locale. A segment naming no routed locale is still not stripped. Nothing to change in an app: prefixed pages take their route's `ttl`, `tags` and purge keys from the first render after the upgrade. A page gated on the clock (`publishedAt <= now`) is busted by no write when the instant passes; it needs a `ttl`, and the `ttl` alone now expires it in every locale.

docs/ops/01-kubernetes.md

@@ -110,6 +110,29 @@ To turn realtime on:
 | 3 | Enable exactly one replicator per database: `roles.replicator.enabled: true` (chart) or a `ROLE=replicator` service (Compose) |
 | 4 | Enable sync: `roles.sync.enabled: true`, or `replicas: 1` on the Compose `sync` service. The chart's Ingress routes `/_x/sync` only while sync is enabled. The node refuses a socket from a page on another ORIGIN — scheme, host and port, compared exactly (`X_SOCKET_ORIGIN_REFUSED`); a socket is refused when its origin is not `APP_URL`'s when one is declared; with none declared, not the origin the node was reached on. With the Ingress enabled the chart sets `APP_URL` on the sync role from `ingress.host` (`https` when `ingress.tls`), and a declared origin is the whole list. `env.APP_URL` wins when set: state the page's origin there when the pages are served on another host than `ingress.host` |
 
+### NATS is a boot dependency of `sync` and the replicator only
+
+`As of 2026-10`. Until then every role awaited the first dial, and the client library retried that
+dial forever: a `web`, `worker` or `scheduler` pod that restarted while NATS was down (an OOM, a
+node drain, a rollout, a scale-up) hung before it bound a socket, so pages, webhooks and jobs went
+down for a bus those roles only publish "re-read" events to.
+
+| Role | NATS down at pod start | NATS down while running | Needs JetStream (`-js`) |
+|---|---|---|---|
+| `sync` | retries for 15 s, then exits with `X_TRANSPORT_UNAVAILABLE`; the kubelet restarts it | not ready (`transport: failing`); reconnects on its own | yes, at boot |
+| `replicator` | same | not ready while disconnected; a publish refused during a restart of a second or two is retried in place, so the stream is not torn down for it | no |
+| `web`, `worker`, `scheduler` | **boots and serves**; dials in the background (jittered, 500 ms doubling to 30 s) | **stays ready**; publishes are refused with `X_TRANSPORT_UNAVAILABLE` until it is back, then resume with no restart | no |
+
+What to watch: `/readyz?deep=1` answers 503 while `checks.transport` is `degraded` — the probe on
+`/readyz` stays 200, by design. The boot line carries the state (`"msg":"ultimate started",…,"bus":"nats(connecting)"`)
+and `"msg":"ultimate bus","bus":"nats(up)"` is logged when the connection lands. While the bus is away a cache
+invalidation still clears the replica that made it, Redis and the CDN; the other replicas' in-process
+copies are stale until it returns. Then each sender publishes the tags it could not (at most 1,024,
+else one flush-all), and every replica that was disconnected empties its in-process `lru` and marks
+its tag-revalidated ISR pages stale — so expect one cold in-process cache per replica after a NATS
+restart, and the database reads that refill it. Do not add NATS to a `web` pod's
+`initContainers` or `dependsOn`: that rebuilds the dependency this removes.
+
 ### Replication is a cluster-wide grant
 
 `REPLICATION` is a role attribute of the whole Postgres **cluster**, not of one database. A

packages/cache/CLAUDE.md

@@ -42,6 +42,9 @@ Tier 1. Tagged caching + THE invalidation graph.
   `TIER_ORDER` alias (`X_HELPER_COPY` refuses a value of core's under a second name). Adding a
   rung is an edit to `packages/core/src/cache-vocabulary.ts` plus a factory here; `scripts/render-modes.ts` refuses a second declaration of the set. **`isr` is
   not a tier** — the `'isr'` in `invalidate.ts` is an ISR route (`DependentKind = 'isr-route'`).
+- **`flushProcessTiers(source)` is for a process that cannot name what it missed** (its bus was
+  away): `CacheTier.clear()` on in-process tiers only, tagged ISR pages stale, fills fenced out
+  (`markAllInvalidated`). Never re-emits. A shared tier must NOT implement `clear`.
 - **`bestEffort()` is the only sanctioned way to swallow a cache refusal.** Its label is `TierLabel`
   (`TierName` plus `'query-read'`), deliberately not a widening of `TierName`.
 - **A refusal is rendered with `renderThrowable()`, never `error.message`** — at all five absorbing

packages/cache/README.md

@@ -304,8 +304,17 @@ bus.subscribe('cache.invalidate', (wireTags) => receiveInvalidationBroadcast(wir
 The inbound half **cannot** re-emit, and that is structural rather than a flag: `emit` lives on the
 private fan-out options and `receiveInvalidationBroadcast` is the only caller that passes `false`.
 A receiver that re-broadcast would be a storm bounded by nothing. A failed send lands in
-`report.errors` under `tier: "broadcast"` — the other pods then clear on TTL, and the write that
-triggered the bust still succeeds. An inbound tag this process has not declared is dropped and
+`report.errors` under `tier: "broadcast"` and the write that triggered the bust still succeeds.
+What happens to that bust is the sender's (`@ultimat3/cli` keeps the refused tags, bounded, and
+publishes them when the bus returns); with no such sender the other pods clear on TTL.
+
+**`flushProcessTiers(source)` is the repair for a process that cannot know what it missed** — its
+bus connection was down for a window, or a peer says it had more refused busts than it could keep.
+It calls `CacheTier.clear()` on every tier that has one (only a tier whose store is this process's
+heap implements it: `lruTier` does, `redisTier` and `cdnTier` do not), marks every ISR page that
+depends on a tag stale through the registered revalidator, and fences out every fill in flight
+(`markAllInvalidated`). It never re-emits and never throws; its report and its `/_x` event carry
+`tags: ['*']` (`FLUSH_ALL_TAG`). An inbound tag this process has not declared is dropped and
 reported rather than thrown, because mid-deploy the new pods know an entity the old ones do not and
 a throw would kill the subscriber loop that delivered it.
 

packages/cache/src/fence.ts

@@ -70,6 +70,17 @@ export function markInvalidated(scope: FenceScope): void {
   }
 }
 
+/**
+ * Everything, invalidated: every fence sampled before this call answers `false`. For a process
+ * that cannot name what it missed (`flushProcessTiers`) — the same "older than the ring remembers:
+ * unprovable, so refused" rule, reached on purpose. One refetch per in-flight fill.
+ */
+export function markAllInvalidated(): void {
+  generation += 1;
+  forgottenThrough = generation;
+  marks.length = 0;
+}
+
 function hits(mark: Mark, scope: FenceScope): boolean {
   if (mark.key !== undefined) return scope.key !== undefined && mark.key === scope.key;
   const owned = mark.tag;

packages/cache/src/flush.test.ts

@@ -0,0 +1,115 @@
+// `flushProcessTiers()`: what a process does when it cannot know which busts it missed. Its bus
+// connection was down for a window nobody measured, so every peer's invalidation in that window is
+// lost to it — and the only correct answer to "which entries are stale" is "any of them".
+
+import { afterAll, beforeEach, describe, expect, test } from 'bun:test';
+import { sampleFence } from './fence';
+import { isolateGraph, registerDependent, resetGraph } from './graph';
+import {
+  flushProcessTiers,
+  isolateTiers,
+  recentInvalidations,
+  registerInvalidationBroadcast,
+  registerRevalidator,
+  registerTier,
+  resetTiers,
+} from './invalidate';
+import { lruTier } from './lru';
+import { isolateDeclaredTags, resetDeclaredTags, tag } from './tags';
+import type { CacheTier } from './tiers';
+
+const restoreRegistries = [isolateTiers(), isolateDeclaredTags(), isolateGraph()];
+
+beforeEach(() => {
+  resetTiers();
+  resetDeclaredTags();
+  resetGraph();
+});
+
+afterAll(() => {
+  for (const restore of restoreRegistries) restore();
+});
+
+/** A tier other processes write too: it was cleared by whoever ran the bust, so it has no `clear`. */
+const sharedTier = (): CacheTier & { cleared: number } => {
+  const tier = {
+    name: 'redis' as const,
+    cleared: 0,
+    get: async () => undefined,
+    set: async () => undefined,
+    del: async () => undefined,
+    invalidateTags: async () => {
+      tier.cleared += 1;
+      return { tier: 'redis' as const, keys: [] };
+    },
+  };
+  return tier;
+};
+
+describe('flushProcessTiers', () => {
+  test('drops every in-process entry — tagged, untagged, whatever a peer may have busted', async () => {
+    const lru = lruTier({ maxBytes: 10_000, defaultTtlMs: 3_600_000 });
+    const redis = sharedTier();
+    registerTier(lru);
+    registerTier(redis);
+    await lru.set('post:1', 'old', { tags: [tag('post', '1')] });
+    await lru.set('feed', ['a'], { tags: [tag('post')] });
+    await lru.set('untagged', 1);
+
+    const report = await flushProcessTiers('cache.bus-reconnect');
+
+    expect(await lru.get('post:1')).toBeUndefined();
+    expect(await lru.get('feed')).toBeUndefined();
+    expect(await lru.get('untagged')).toBeUndefined();
+    // The shared tier is not this process's to empty: the bust's sender already cleared it.
+    expect(redis.cleared).toBe(0);
+    expect(report.tags).toEqual(['*']);
+    expect(report.tiers.map((entry) => entry.tier)).toEqual(['lru']);
+    expect(report.errors).toEqual([]);
+  });
+
+  test('marks every tag-revalidated ISR page stale: a peer bust reaches them by no other road', async () => {
+    registerDependent([tag('post')], { kind: 'isr-route', id: '/blog' });
+    registerDependent([tag('post', '1'), tag('author', '9')], { kind: 'isr-route', id: '/blog/1' });
+    registerDependent([tag('post')], { kind: 'live-query', id: 'q1' });
+    const marked: string[] = [];
+    registerRevalidator((path) => {
+      marked.push(path);
+    });
+
+    const report = await flushProcessTiers('cache.bus-reconnect');
+
+    expect([...marked].sort()).toEqual(['/blog', '/blog/1']);
+    expect([...report.isr].sort()).toEqual(['/blog', '/blog/1']);
+  });
+
+  test('a fill that was already loading cannot republish what the flush dropped', async () => {
+    const fence = sampleFence({ key: 'post:1', tags: [tag('post', '1')] });
+    expect(fence.isValid()).toBe(true);
+
+    await flushProcessTiers('cache.bus-reconnect');
+
+    expect(fence.isValid()).toBe(false);
+    // One taken afterwards is unaffected: the flush is an instant, not a mode.
+    expect(sampleFence({ key: 'post:1' }).isValid()).toBe(true);
+  });
+
+  test('never re-emits, is recorded under its source, and survives a tier that will not clear', async () => {
+    let sends = 0;
+    registerInvalidationBroadcast(() => {
+      sends += 1;
+    });
+    const lru = lruTier({ maxBytes: 10_000, defaultTtlMs: 3_600_000 });
+    registerTier({
+      ...lru,
+      clear: () => Promise.reject(new RangeError('the tier refused')),
+    });
+
+    const report = await flushProcessTiers('cache.broadcast');
+
+    expect(sends).toBe(0);
+    expect(report.errors).toEqual([{ tier: 'lru', message: expect.stringContaining('refused') }]);
+    expect(recentInvalidations()[0]?.source).toBe('cache.broadcast');
+    expect(recentInvalidations()[0]?.tags).toEqual(['*']);
+  });
+});

packages/cache/src/index.ts

@@ -41,6 +41,8 @@ export type {
   Revalidator,
 } from './invalidate';
 export {
+  FLUSH_ALL_TAG,
+  flushProcessTiers,
   invalidateTags,
   invalidateWireTags,
   isolateTiers,