Recover Rip access logs quietly after Janus restarts

packages/sites/README.md

@@ -1269,6 +1269,11 @@ passed. You can edit `access.format` while the site is running; restart the
 manager (`rip sites restart <app>`, or stop/start a foreground `rip sites run`) to
 pick up the new picture — the access stream does not hot-reload `serve.rip`.
 
+When Janus restarts, the manager re-registers its app and moves access logging
+to the replacement ID. A stream that receives 404 waits for that registration
+instead of retrying the deleted ID. Repeated connection failures are reported
+once until the stream reconnects; a changed error is still reported.
+
 `app.root` selects the browser App directory relative to the project.
 `app.changes` classifies authored files by client apply verdict. The block
 shown above is the complete default; omitting `app`, `changes`, or one of its

packages/sites/control.rip

@@ -2,6 +2,12 @@ import { existsSync, lstatSync, mkdirSync, realpathSync, statSync } from 'node:f
 import { dirname, join, resolve } from 'node:path'
 import { tmpdir } from 'node:os'
 
+export janusErrorMessage = (error) ->
+  if error?.code in ['FailedToOpenSocket', 'ConnectionRefused', 'ECONNREFUSED', 'ECONNRESET']
+    'Janus unavailable'
+  else
+    "#{error?.message or error}"
+
 # Project identity and private manager discovery are shared by foreground
 # commands and the per-user agent. One canonical root must
 # always select one manager socket, regardless of which client asks.

packages/sites/manager.rip

@@ -6,7 +6,7 @@ import { brotliCompressSync, brotliDecompressSync, constants as zlib } from 'nod
 import { rash, check } from 'rip/app/rash'
 import { packageInventory } from './bundle.rip'
 import { AccessClient, parseAccessPicture, pumpAccessOutput, writeAccessOutput } from './monitor.rip'
-import { managerRequest, managerRuntime, resolveProject, waitManagerStopped } from './control.rip'
+import { janusErrorMessage, managerRequest, managerRuntime, resolveProject, waitManagerStopped } from './control.rip'
 import { configureEdge, edgeControl, edgeFetch } from './edge.rip'
 
 positiveEnv = (name, fallback) ->
@@ -559,7 +559,7 @@ runBrowse = (args) ->
         else unless response.ok
           warn "rip-sites: browse heartbeat failed (#{response.status})"
       catch error
-        warn "rip-sites: browse heartbeat failed: #{error?.message or error}" unless stopping
+        warn "rip-sites: browse heartbeat failed: #{janusErrorMessage(error)}" unless stopping
       finally
         heartbeatTask = null
     heartbeatTask
@@ -2167,7 +2167,7 @@ export main = (argv) ->
         else unless response.ok
           warn "rip-sites: Janus heartbeat failed (#{response.status})"
       catch error
-        warn "rip-sites: Janus heartbeat failed: #{String(error)}" unless stopping
+        warn "rip-sites: Janus heartbeat failed: #{janusErrorMessage(error)}" unless stopping
       finally
         heartbeatTask = null
     heartbeatTask

packages/sites/monitor.rip

@@ -1,4 +1,5 @@
 import { identifierRuns, isIdentifierName } from '../../src/ident.js'
+import { janusErrorMessage } from './control.rip'
 
 export DEFAULT_ACCESS_PICTURE =! '{local_time} {local_timezone} {duration_seconds:@s} │ {status} {mime_abbrev:<4} {response_bytes:@B} │ {method} {path} │ {mark}'
 export MAX_ACCESS_LINE_BYTES =! 8192
@@ -689,14 +690,17 @@ export class AccessClient
   run: (generation, appId, signal) ->
     cursor = '0'
     base = 100
+    reportedError = null
     while generation is @generation and not signal.aborted
       validHello = false
+      missingRegistration = false
       reader = null
       try
         requested = cursor
         response = @request! appId, requested, signal
         unless response.status is 200
           detail = (response.text!).trim()
+          missingRegistration = response.status is 404
           throw Error.new("access stream HTTP #{response.status}#{if detail then ": #{detail}" else ''}")
         throw Error.new('access stream response has no body') unless response.body?
         reader = response.body.getReader()
@@ -722,6 +726,7 @@ export class AccessClient
               throw Error.new('hello does not match registration generation') unless record.app_id is appId and record.after is requested
               validHello = true
               base = 100
+              reportedError = null
               helloHead = record.head
               highWater = record.head
               expectedInitialGap = compareDecimal(record.head, requested) > 0
@@ -768,10 +773,17 @@ export class AccessClient
         e = caught
         try reader?.cancel().catch(-> null)
         throw e if e?.accessOutput or e?.code is 'EPIPE'
-        return if signal.aborted
-        try writeAccessOutput!(@stderr, "rip-sites: access stream: #{e?.message or e}; reconnecting\n")
-        catch outputError
-          throw outputError
+        return if signal.aborted or generation isnt @generation
+        message = if missingRegistration
+          "app #{appId} is no longer registered; waiting for app re-registration"
+        else
+          "#{janusErrorMessage(e)}; reconnecting"
+        if message isnt reportedError
+          writeAccessOutput! @stderr, "rip-sites: access stream: #{message}\n"
+          reportedError = message
+        # Only the manager can replace a registration. Retrying its dead
+        # ID cannot recover the stream; move starts the new subscription.
+        return if missingRegistration
       break unless generation is @generation and not signal.aborted
       delay = Math.floor(@sample() * (base + 1))
       try

packages/sites/test/janus/test.rip

@@ -100,13 +100,15 @@ certification = ->
 
   caddy = null
   manager = null
-  try
-    caddy = Bun.spawn [CADDY, 'run', '--config', config, '--adapter', 'caddyfile'],
+  startEdge = ->
+    Bun.spawn [CADDY, 'run', '--config', config, '--adapter', 'caddyfile'],
       cwd: dir
       stdin: 'ignore'
       stdout: 'pipe'
       stderr: 'pipe'
       env: { ...process.env, JANUS_HEARTBEAT_TTL: '6s', XDG_DATA_HOME: join(dir, 'data'), XDG_CONFIG_HOME: join(dir, 'config') }
+  try
+    caddy = startEdge()
     caddyOut = Response.new(caddy.stdout).text()
     caddyErr = Response.new(caddy.stderr).text()
     waitFor! 'control socket', ->
@@ -121,7 +123,16 @@ certification = ->
       stdin: 'ignore'
       stdout: 'pipe'
       stderr: 'pipe'
-    managerOut = Response.new(manager.stdout).text()
+    managerText = ''
+    managerOut = do ->
+      outputReader = manager.stdout.getReader()
+      decoder = TextDecoder.new()
+      loop
+        { done, value } = outputReader.read!
+        break if done
+        managerText += decoder.decode(value, { stream: true })
+      managerText += decoder.decode()
+      managerText
     managerErr = Response.new(manager.stderr).text()
 
     edgeDiagnostic = 'no response'
@@ -224,9 +235,34 @@ certification = ->
       catch then null
     eq second, { version: 'two' }
 
+    caddy.kill 'SIGTERM'
+    await caddy.exited
+    # Keep the edge down through a heartbeat to exercise both control
+    # clients' connection diagnostics before registration recovery.
+    sleep! 5100
+    caddy = startEdge()
+    caddyOut = Response.new(caddy.stdout).text()
+    caddyErr = Response.new(caddy.stderr).text()
+    waitFor! 'registration after edge restart', (->
+      try
+        registered = (fetch!('http://janus/1.0/apps', { unix: control })).json!
+        registered.length is 1 and registered[0].id isnt apps[0].id
+      catch then false
+    ), 15000
+    recovered = edgeFetch! port, '/api/version'
+    eq recovered.json!, { version: 'two' }
+    # A path requested only after restart proves the manager's access
+    # subscription follows the replacement registration, too.
+    observed = edgeFetch! port, '/api/restart-observation'
+    observed.text!
+    waitFor! 'access output after edge restart', -> managerText.includes('/api/restart-observation')
+
     manager.kill 'SIGTERM'
     eq (await manager.exited), 143, await managerErr
     ok (await managerOut).includes('/api/version'), 'released Janus access events must reach foreground output'
+    ok (await managerOut).includes('/api/restart-observation'), 'access output must resume after edge restart'
+    ok (await managerErr).includes('Janus heartbeat failed: Janus unavailable')
+    ok not (await managerErr).includes('Was there a typo')
     manager = null
     waitFor! 'deregistration', ->
       response = edgeFetch! port, '/api/version'

packages/sites/test/monitor/test.rip

@@ -908,3 +908,54 @@ test! 'foreground raw stdout is pure NDJSON and off opens no subscription', ->
   disabled = run! 'off'
   eq disabled.accessCount, 0
   ok disabled.output.includes 'rip-sites: https://cart.test/'
+
+test! 'a missing registration waits for move instead of retrying its dead ID', ->
+  calls = []
+  output = []
+  diagnostic = []
+  fetcher = (url, options) ->
+    calls.push URL.new(url).pathname
+    return Response.json({ error: 'unknown app id' }, { status: 404 }) if calls.at(-1).includes('old-aaaaaa')
+    Response.new ReadableStream.new
+      start: (controller) ->
+        controller.enqueue helloLine()
+        controller.enqueue accessLine(sequence: '1')
+        options.signal.addEventListener 'abort', (-> controller.close()), { once: true }
+  client = AccessClient.new { mode: 'raw', control: { base: 'http://janus' }, stdout: memorySink(output), stderr: memorySink(diagnostic), fetch: fetcher, sample: -> 0 }
+  try
+    client.move 'old-aaaaaa'
+    waitFor! -> calls.length > 0
+    sleep! 30
+    eq calls, ['/1.0/apps/old-aaaaaa/access']
+    ok Buffer.concat(diagnostic).toString().includes('waiting for app re-registration')
+    client.move 'cart-abc123'
+    waitFor! -> output.length is 2
+    eq calls, ['/1.0/apps/old-aaaaaa/access', '/1.0/apps/cart-abc123/access']
+    eq JSON.parse(Buffer.from(output[1]).toString()).sequence, '1'
+  finally
+    client.stop!
+
+test! 'connection outages report once and a recovered stream resets the diagnostic', ->
+  diagnostic = []
+  calls = 0
+  fetcher = (_url, options) ->
+    calls++
+    if calls <= 3 or calls is 5
+      throw Object.assign(Error.new('Was there a typo in the url or port?'), { code: 'FailedToOpenSocket' })
+    if calls is 4
+      return Response.new ReadableStream.new
+        start: (controller) ->
+          controller.enqueue helloLine()
+          controller.close()
+    Promise.new (_resolve, reject) ->
+      options.signal.addEventListener 'abort', (-> reject(options.signal.reason)), { once: true }
+  client = AccessClient.new { mode: 'raw', control: { base: 'http://janus' }, stdout: memorySink([]), stderr: memorySink(diagnostic), fetch: fetcher, sleep: -> null }
+  try
+    client.move 'cart-abc123'
+    waitFor! -> calls is 6
+    text = Buffer.concat(diagnostic).toString()
+    eq text.trim().split('\n').length, 2
+    ok text.includes('Janus unavailable; reconnecting')
+    ok not text.includes('typo')
+  finally
+    client.stop!