diff --git a/packages/sites/README.md b/packages/sites/README.md index 9c26d7e2..b392f1c1 100644 --- a/packages/sites/README.md +++ b/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 `, 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 diff --git a/packages/sites/control.rip b/packages/sites/control.rip index 7258faf5..9d474e41 100644 --- a/packages/sites/control.rip +++ b/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. diff --git a/packages/sites/manager.rip b/packages/sites/manager.rip index 9800f88a..6bf4b3cc 100644 --- a/packages/sites/manager.rip +++ b/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 diff --git a/packages/sites/monitor.rip b/packages/sites/monitor.rip index 1e9e2623..e979695e 100644 --- a/packages/sites/monitor.rip +++ b/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 diff --git a/packages/sites/test/janus/test.rip b/packages/sites/test/janus/test.rip index 3213e6f2..c48f98da 100644 --- a/packages/sites/test/janus/test.rip +++ b/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' diff --git a/packages/sites/test/monitor/test.rip b/packages/sites/test/monitor/test.rip index cff4edef..9928826d 100644 --- a/packages/sites/test/monitor/test.rip +++ b/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!