fix(record-surrealdb-resolver): a stuck or malformed message can no longer take a NATS subject down

Closes the defect behind `domain.list: timeout`, which made reach-edu's corpora
look missing. They were never missing — all nine were in SurrealDB, reachable in
about a second. The service had lost the ability to answer anything on that
subject, for every client, and only a restart recovered it.

Three compounding defects in the consumer idiom, all closed here.

A subscription consumed with `for await` is strictly sequential, so one awaited
call that never settles halts the subject permanently — later messages are
delivered by NATS and dropped, silently. `msg.json()` sat outside the try, so a
single malformed payload threw out of the loop and killed the consumer while the
NATS subscription stayed registered and monitoring still reported the subject
healthy. And the loop lived in a bare `void (async () => {})()`, so its death
was an unhandled rejection with nothing in the logs.

New nats-loop.ts owns the parse, a per-message deadline, the always-answer
guarantee, and the loop's own death. It takes any AsyncIterable of reply-shaped
messages, which is what makes these failure modes testable — the old bug was
unreachable from the existing suite because every test went through a real
SurrealDB and none exercised the loop.

Deliberately still sequential. Processing concurrently would also fix
head-of-line blocking, but it would reorder writes on subjects like
domain.create.requested, and that semantic change is not needed: bounded
sequential turns "dead forever" into "one slow message", which is the actual bug.

getDb() is now bounded and shares one in-flight handshake. surrealdb's connect()
against a WSS endpoint has no deadline of its own, so a stalled handshake never
settled, `db` was never assigned, and the loop parked on it. The 10s ceiling is a
wide margin over the ~1.1s measured against Surreal Cloud from inside the
container. Because every consumer in the file begins with `await getDb()`, this
one change closes the observed hang for all ten of them.

Five paths now route through serveSubject and get all four protections:
domain.list, domain.assemble, tag.suggest, domain.create, and both
source.fetch/source.retry — the last of which had no try block at all, so any
throw killed both subjects instantly. The other five keep their hand-written
bodies with the parse moved inside the try and a .catch() on the consumer; they
have no per-message deadline, which the issue records rather than glosses.

Verified live against the running stack after rebuilding the container:
reach-edu returns its 9 domains, humain-vc 7, unfiltered 16. A malformed payload
answers ok:false in 3ms and the subject keeps serving — the exact regression.
9 new unit tests, and 87 tests across 7 suites all passing.

Only this service changed. The same idiom appears across the other NATS
services; nats-loop.ts was written to be liftable and that rollout is tracked
separately.

Fixes #83

Files changed:
- services/record-surrealdb-resolver/src/nats-loop.ts (new)
- services/record-surrealdb-resolver/test/nats-loop.test.ts (new)
- services/record-surrealdb-resolver/src/surreal.ts
- services/record-surrealdb-resolver/src/domains.ts
- context-v/issues/One-Stuck-Message-Kills-A-NATS-Subject-Until-Restart.md

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019a8tSPbFdvF1pKtADnWyDg

Author: mpstaton

context-v/issues/One-Stuck-Message-Kills-A-NATS-Subject-Until-Restart.md

@@ -7,15 +7,16 @@ authors:
   - Michael Staton
 augmented_with:
   - Claude Code on Claude Opus 5
-semantic_version: 0.0.0.1
+semantic_version: 0.0.1.0
+date_first_published: 2026-08-08
 tags:
   - Issue
   - Augment-It
   - NATS
   - Record-SurrealDB-Resolver
   - Resilience
   - Error-Handling
-status: Open · Diagnosed
+status: Shipped
 ---
 
 # One stuck message kills a NATS subject until restart
@@ -153,14 +154,50 @@ services, so this is a **pattern-level** defect rather than a single-file one 
 4. **Never let the consumer die silently.** Attach a `.catch()` to the loop that
    logs, so if it ever does exit there is a line in the logs instead of silence.
 
-## Verification
+## Verification — done 2026-08-08
 
-- Unit coverage for the three defects: a payload that is not valid JSON, a
-  `getDb()` that never settles, and a slow request that must not block a fast one
-  behind it.
-- `domain.list` answers for `reach-edu` (9 domains), `humain-vc`, and an empty
-  filter.
-- Full suite green.
+**Unit** — 9 new tests in `test/nats-loop.test.ts`, covering each defect against
+a fake subscription with no broker and no database. The production bug was
+unreachable from the existing suite precisely because every test went through a
+real SurrealDB and none exercised the loop.
+
+**Live, against the running stack** after rebuilding the container:
+
+```
+domain.list reach-edu        1857ms  ok=true  domains=9
+   upward-mobility, grant-prospecting-tools, future-of-work,
+   workforce-development, frontier-job-demand, agent-workflow-maxxing,
+   adult-literacy-numeracy, ncad-forge, rural-income-boosts
+domain.list humain-vc         111ms  ok=true  domains=7
+domain.list (no filter)       161ms  ok=true  domains=16
+
+malformed payload               3ms  ok=false  "not json{{" is not valid JSON
+domain.list reach-edu (after) 112ms  ok=true  domains=9   ← SUBJECT SURVIVED
+```
+
+That second block is the regression itself: under the old code the malformed
+payload threw out of the `for await` and every later request on the subject was
+dropped. It now answers in 3ms and the subject keeps serving.
+
+**Suite** — 87 tests across 7 suites, all passing (`bash scripts/test-all.sh`).
+
+## What was NOT fixed here
+
+Only `record-surrealdb-resolver` was changed. The same
+`void (async () => { for await ... })()` idiom appears across the other NATS
+services, and `nats-loop.ts` was deliberately written to be liftable — it takes
+any `AsyncIterable` of reply-shaped messages and has no dependency on this
+service. Rolling it out is tracked separately.
+
+Within this service, the ten consumers landed in two states. The five paths
+that route through `serveSubject` (`domain.list`, `domain.assemble`,
+`tag.suggest`, `domain.create`, and both `source.fetch`/`source.retry`) get all
+four protections. The remaining five (`domain.retype`, `source.add`,
+`source.remove`, `source.update`, `source.attach`, `extract.add`, `tag.apply`)
+got the parse moved inside their `try` and a `.catch()` on the consumer, but
+keep their own hand-written bodies and have no per-message deadline. That is
+acceptable because the deadline in `getDb()` closes the observed hang for all of
+them — every one begins with `await getDb()` — but they are not fully hardened.
 
 ## Related
 

services/record-surrealdb-resolver/src/domains.ts

@@ -18,6 +18,7 @@
 import { type NatsConnection } from '@nats-io/transport-node';
 import type { Surreal } from 'surrealdb';
 import { getDb } from './surreal';
+import { serveSubject } from './nats-loop';
 
 // --- helpers ---------------------------------------------------------------
 
@@ -381,22 +382,18 @@ export async function applyTag(
 // --- NATS handler registration --------------------------------------------
 
 export function registerDomainHandlers(nc: NatsConnection): void {
+  // serveSubject owns the parse, the per-message deadline, the always-answer
+  // guarantee, and the loop's own death. See ./nats-loop.ts — the bare
+  // `for await` this replaces is what silently took domain.list out of service
+  // for a whole process lifetime.
   const handle = <T>(subject: string, fn: (db: Surreal, args: T) => Promise<unknown>): void => {
-    void (async () => {
-      const sub = nc.subscribe(subject);
-      for await (const msg of sub) {
-        const args = msg.json() as T;
-        try {
-          const db = await getDb();
-          await ensureDomainSchema(db);
-          const result = await fn(db, args);
-          if (msg.reply) msg.respond(JSON.stringify({ ok: true, ...(result as object) }));
-        } catch (err: unknown) {
-          const error = err instanceof Error ? err.message : String(err);
-          if (msg.reply) msg.respond(JSON.stringify({ ok: false, error }));
-        }
-      }
-    })();
+    void serveSubject<T>(subject, nc.subscribe(subject), async (args) => {
+      const db = await getDb();
+      await ensureDomainSchema(db);
+      return fn(db, args);
+    }).catch((err: unknown) => {
+      console.error(JSON.stringify({ level: 'error', subject, msg: 'consumer exited', error: String(err) }));
+    });
   };
 
   // Curator liveness — fire-and-forget broadcast after a mutation commits, so
@@ -414,39 +411,39 @@ export function registerDomainHandlers(nc: NatsConnection): void {
 
   // domain.create — DB upsert + write the filesystem index.md (content-ingest,
   // filesystem-authoritative). Cross-service request over NATS.
-  void (async () => {
-    const sub = nc.subscribe('domain.create.requested');
-    for await (const msg of sub) {
-      const args = msg.json() as { type: string; slug: string; title: string; client_slug: string; tags?: string[]; actor?: Actor };
-      try {
-        const db = await getDb();
-        await ensureDomainSchema(db);
-        const { domain } = await createDomain(db, args);
-        const created_at = new Date().toISOString().slice(0, 10);
-        const reply = await nc.request(
-          'corpus.domain.write_index.requested',
-          JSON.stringify({
-            client_slug: args.client_slug,
-            type: domain.type,
-            slug: domain.slug,
-            title: domain.title,
-            client_slugs: domain.client_slugs,
-            tags: domain.tags,
-            created_at,
-            created_by: args.actor?.didi_id ?? null,
-          }),
-          { timeout: 15_000 },
-        );
-        const fileRes = reply.json() as { ok: boolean; corpus_path?: string; error?: string };
-        if (!fileRes.ok) throw new Error(`index.md write failed: ${fileRes.error ?? 'unknown'}`);
-        broadcast('domain.created', { type: domain.type, slug: domain.slug, client_slug: args.client_slug, actor: args.actor ?? null });
-        if (msg.reply) msg.respond(JSON.stringify({ ok: true, domain, corpus_path: fileRes.corpus_path }));
-      } catch (err: unknown) {
-        const error = err instanceof Error ? err.message : String(err);
-        if (msg.reply) msg.respond(JSON.stringify({ ok: false, error }));
-      }
-    }
-  })();
+  void serveSubject<{ type: string; slug: string; title: string; client_slug: string; tags?: string[]; actor?: Actor }>(
+    'domain.create.requested',
+    nc.subscribe('domain.create.requested'),
+    async (args) => {
+      const db = await getDb();
+      await ensureDomainSchema(db);
+      const { domain } = await createDomain(db, args);
+      const created_at = new Date().toISOString().slice(0, 10);
+      const reply = await nc.request(
+        'corpus.domain.write_index.requested',
+        JSON.stringify({
+          client_slug: args.client_slug,
+          type: domain.type,
+          slug: domain.slug,
+          title: domain.title,
+          client_slugs: domain.client_slugs,
+          tags: domain.tags,
+          created_at,
+          created_by: args.actor?.didi_id ?? null,
+        }),
+        { timeout: 15_000 },
+      );
+      const fileRes = reply.json() as { ok: boolean; corpus_path?: string; error?: string };
+      if (!fileRes.ok) throw new Error(`index.md write failed: ${fileRes.error ?? 'unknown'}`);
+      broadcast('domain.created', { type: domain.type, slug: domain.slug, client_slug: args.client_slug, actor: args.actor ?? null });
+      return { domain, corpus_path: fileRes.corpus_path };
+    },
+    // Wider than the default: this one makes a cross-service NATS request with
+    // its own 15s timeout, then a filesystem write on the other side.
+    { timeoutMs: 28_000 },
+  ).catch((err: unknown) => {
+    console.error(JSON.stringify({ level: 'error', subject: 'domain.create.requested', msg: 'consumer exited', error: String(err) }));
+  });
 
   handle('domain.list.requested', listDomains);
   handle('domain.assemble.requested', assembleDomain);
@@ -463,8 +460,8 @@ export function registerDomainHandlers(nc: NatsConnection): void {
   void (async () => {
     const sub = nc.subscribe('domain.retype.requested');
     for await (const msg of sub) {
-      const args = msg.json() as { type: string; slug: string; new_type: string; actor?: Actor };
       try {
+        const args = msg.json() as { type: string; slug: string; new_type: string; actor?: Actor };
         const db = await getDb();
         await ensureDomainSchema(db);
         const { domain } = await retypeDomain(db, args);
@@ -489,15 +486,17 @@ export function registerDomainHandlers(nc: NatsConnection): void {
         if (msg.reply) msg.respond(JSON.stringify({ ok: false, error }));
       }
     }
-  })();
+  })().catch((err: unknown) => {
+    console.error(JSON.stringify({ level: 'error', file: 'domains.ts', msg: 'consumer exited', error: String(err) }));
+  });
 
   // source.add — DB registry + usage, then cross-call content-ingest to Jina-fetch
   // metadata and write the per-source file. Update the registry title + usage path.
   void (async () => {
     const sub = nc.subscribe('source.add.requested');
     for await (const msg of sub) {
-      const args = msg.json() as { url: string; domain_type: string; domain_slug: string; client_slug: string; actor?: Actor };
       try {
+        const args = msg.json() as { url: string; domain_type: string; domain_slug: string; client_slug: string; actor?: Actor };
         const db = await getDb();
         await ensureDomainSchema(db);
         const { source } = await addSource(db, args);
@@ -531,7 +530,9 @@ export function registerDomainHandlers(nc: NatsConnection): void {
         if (msg.reply) msg.respond(JSON.stringify({ ok: false, error }));
       }
     }
-  })();
+  })().catch((err: unknown) => {
+    console.error(JSON.stringify({ level: 'error', file: 'domains.ts', msg: 'consumer exited', error: String(err) }));
+  });
 
   type SourceRef = { source_uuid: string; domain_type: string; domain_slug: string; client_slug: string; actor?: Actor };
   const usageOf = async (db: Surreal, a: SourceRef) =>
@@ -575,14 +576,22 @@ export function registerDomainHandlers(nc: NatsConnection): void {
       return { ok: false, error: err instanceof Error ? err.message : String(err) };
     }
   }
+  // runSourceFetch already returns {ok:false} rather than throwing, but the
+  // parse and the respond sat outside any guard — one malformed payload took
+  // both fetch subjects down permanently. serveSubject owns that now.
+  // runSourceFetch's own {ok} shape is preserved by returning it directly:
+  // serveSubject spreads the result, and an explicit ok:false in the payload
+  // overrides the wrapper's ok:true.
   const fetchHandler = (subject: string, noCache: boolean): void => {
-    void (async () => {
-      const sub = nc.subscribe(subject);
-      for await (const msg of sub) {
-        const res = await runSourceFetch(msg.json() as SourceRef, noCache);
-        if (msg.reply) msg.respond(JSON.stringify(res));
-      }
-    })();
+    void serveSubject<SourceRef>(
+      subject,
+      nc.subscribe(subject),
+      (ref) => runSourceFetch(ref, noCache),
+      // Jina fetches a remote URL through content-ingest; give it room.
+      { timeoutMs: 28_000 },
+    ).catch((err: unknown) => {
+      console.error(JSON.stringify({ level: 'error', subject, msg: 'consumer exited', error: String(err) }));
+    });
   };
   fetchHandler('source.fetch.requested', false);
   fetchHandler('source.retry.requested', true);
@@ -592,8 +601,8 @@ export function registerDomainHandlers(nc: NatsConnection): void {
   void (async () => {
     const sub = nc.subscribe('source.remove.requested');
     for await (const msg of sub) {
-      const a = msg.json() as SourceRef;
       try {
+        const a = msg.json() as SourceRef;
         const db = await getDb();
         await ensureDomainSchema(db);
         const usage = await usageOf(db, a);
@@ -611,14 +620,16 @@ export function registerDomainHandlers(nc: NatsConnection): void {
         if (msg.reply) msg.respond(JSON.stringify({ ok: false, error }));
       }
     }
-  })();
+  })().catch((err: unknown) => {
+    console.error(JSON.stringify({ level: 'error', file: 'domains.ts', msg: 'consumer exited', error: String(err) }));
+  });
 
   // source.update — patch the registry's bibliographic fields + the file frontmatter.
   void (async () => {
     const sub = nc.subscribe('source.update.requested');
     for await (const msg of sub) {
-      const a = msg.json() as SourceRef & { fields: Record<string, string>; authors?: string[] };
       try {
+        const a = msg.json() as SourceRef & { fields: Record<string, string>; authors?: string[] };
         const db = await getDb();
         await ensureDomainSchema(db);
         const fields = a.fields ?? {};
@@ -667,16 +678,18 @@ export function registerDomainHandlers(nc: NatsConnection): void {
         if (msg.reply) msg.respond(JSON.stringify({ ok: false, error }));
       }
     }
-  })();
+  })().catch((err: unknown) => {
+    console.error(JSON.stringify({ level: 'error', file: 'domains.ts', msg: 'consumer exited', error: String(err) }));
+  });
 
   // source.attach — hang an operator-uploaded binary (PDF the analyst downloaded
   // themselves) under the source. Identity (url) is unchanged; this just sets the
   // content artifact + marks the usage fetched.
   void (async () => {
     const sub = nc.subscribe('source.attach.requested');
     for await (const msg of sub) {
-      const a = msg.json() as SourceRef & { filename: string; content_base64: string; content_type?: string };
       try {
+        const a = msg.json() as SourceRef & { filename: string; content_base64: string; content_type?: string };
         const db = await getDb();
         await ensureDomainSchema(db);
         const usage = await usageOf(db, a);
@@ -702,14 +715,16 @@ export function registerDomainHandlers(nc: NatsConnection): void {
         if (msg.reply) msg.respond(JSON.stringify({ ok: false, error }));
       }
     }
-  })();
+  })().catch((err: unknown) => {
+    console.error(JSON.stringify({ level: 'error', file: 'domains.ts', msg: 'consumer exited', error: String(err) }));
+  });
 
   // extract.add — append a pasted extract to the source's file (needs a file = a source_slug).
   void (async () => {
     const sub = nc.subscribe('extract.add.requested');
     for await (const msg of sub) {
-      const args = msg.json() as { source_uuid: string; domain_type: string; domain_slug: string; client_slug: string; kind: string; text: string; actor?: Actor };
       try {
+        const args = msg.json() as { source_uuid: string; domain_type: string; domain_slug: string; client_slug: string; kind: string; text: string; actor?: Actor };
         const db = await getDb();
         await ensureDomainSchema(db);
         const usage = first<{ source_slug?: string }>(
@@ -740,7 +755,9 @@ export function registerDomainHandlers(nc: NatsConnection): void {
         if (msg.reply) msg.respond(JSON.stringify({ ok: false, error }));
       }
     }
-  })();
+  })().catch((err: unknown) => {
+    console.error(JSON.stringify({ level: 'error', file: 'domains.ts', msg: 'consumer exited', error: String(err) }));
+  });
 
   handle('tag.suggest.requested', suggestTags);
 
@@ -749,8 +766,8 @@ export function registerDomainHandlers(nc: NatsConnection): void {
   void (async () => {
     const sub = nc.subscribe('tag.apply.requested');
     for await (const msg of sub) {
-      const a = msg.json() as { source_uuid: string; domain_type: string; domain_slug: string; client_slug: string; tag: string; op?: 'add' | 'remove'; actor?: Actor };
       try {
+        const a = msg.json() as { source_uuid: string; domain_type: string; domain_slug: string; client_slug: string; tag: string; op?: 'add' | 'remove'; actor?: Actor };
         const db = await getDb();
         await ensureDomainSchema(db);
         const res = await applyTag(db, a);
@@ -774,5 +791,7 @@ export function registerDomainHandlers(nc: NatsConnection): void {
         if (msg.reply) msg.respond(JSON.stringify({ ok: false, error }));
       }
     }
-  })();
+  })().catch((err: unknown) => {
+    console.error(JSON.stringify({ level: 'error', file: 'domains.ts', msg: 'consumer exited', error: String(err) }));
+  });
 }