Reactor Core: signals, event bus, job scheduler, and coordinator kernel
Este script não deve ser instalado diretamente. É uma biblioteca destinada a ser incluída por outros scripts através da diretiva de metadados // @require https://update.greasyfork.org/scripts/591444/1909485/Brazen%20Framework%20-%20Reactor.js
// ==UserScript==
// @name Brazen Framework - Reactor
// @namespace brazenvoid
// @version 1.0.0
// @author brazenvoid
// @license GPL-3.0-only
// @description Reactor Core: signals, event bus, job scheduler, and coordinator kernel
// @run-at document-end
// ==/UserScript==
// @ts-nocheck
// -------------------------------------------------------------------------
// ReactorProfiler
// -------------------------------------------------------------------------
const BRAZEN_REACTOR_PROFILE_STORAGE_KEY = 'brazen:reactor:profile'
const BRAZEN_REACTOR_INCIDENT_STORAGE_KEY = 'brazen:reactor:last-incident'
const BRAZEN_REACTOR_RING_CAP = 2000
const BRAZEN_REACTOR_INCIDENT_ROWS = 120
const BRAZEN_REACTOR_LONG_TASK_MS = 50
const BRAZEN_REACTOR_WARN_COOLDOWN_MS = 5000
const BRAZEN_REACTOR_PERSIST_DEBOUNCE_MS = 750
const BRAZEN_REACTOR_CAUSE_CAP = 512
const BRAZEN_REACTOR_PATH_FANOUT_CAP = 128
const BRAZEN_REACTOR_IDLE_QUIET_MS = 750
const BRAZEN_REACTOR_IDLE_WINDOW_MS = 10000
const BRAZEN_REACTOR_DIAG_CAP = 500
const BRAZEN_REACTOR_CORE_CHANNELS = new Set(['signals', 'scheduler', 'kernel', 'lane', 'effect', 'framework', 'boot'])
/**
* @typedef {object} BrazenReactorProfileRow
* @property {number} t
* @property {string} channel
* @property {string} event
* @property {string} [key]
* @property {unknown} [data]
* @property {number} [ms]
* @property {number} [causeId]
*/
class BrazenReactorProfiler
{
/** @type {BrazenReactorProfiler|null} */
static _instance = null
/** @return {BrazenReactorProfiler} */
static get()
{
if (!BrazenReactorProfiler._instance) {
BrazenReactorProfiler._instance = new BrazenReactorProfiler()
}
return BrazenReactorProfiler._instance
}
/** @type {boolean} */
disabled = false
/** @type {boolean} */
verbose = false
/** @type {BrazenReactorProfileRow[]} */
_ring = []
/** @type {Map<string, number>} */
_depth = new Map()
/** @type {Map<string, {count: number, totalMs: number, maxMs: number, lastT: number, suspected: boolean}>} */
_stats = new Map()
/** @type {Map<string, number[]>} */
_fireTimes = new Map()
/** @type {Map<string, number>} */
_lastWarnAt = new Map()
/** @type {Map<string, number>} */
_breakerCounts = new Map()
/** @type {Map<string, number>} */
_cooldownUntil = new Map()
/** @type {Map<string, number>} */
_writeStormCounts = new Map()
/** @type {number|null} */
_writeStormMacrotask = null
/** @type {Map<string, {rootKey: string, depth: number}>} */
_spawnContext = new Map()
/** @type {object|null} */
lastIncident = null
/** @type {ReturnType<typeof setTimeout>|null} */
_persistTimer = null
/** @type {boolean} */
_lifecycleBound = false
/** @type {number} */
_causeSeq = 0
/** @type {number[]} */
_causeStack = []
/** @type {Map<number, {causeId: number, parentId: number|null, kind: string, label: string, detail: unknown, rowCount: number, totalMs: number, reentrantCount: number}>} */
_causes = new Map()
/** @type {Map<string, number>} */
_reentrantCounts = new Map()
/** @type {Map<string, number>} */
_scheduledReentrantCounts = new Map()
/** @type {Map<string, Set<string>>} */
_scheduledReentrantSamples = new Map()
/** @type {Map<string, number>} */
_pathFanout = new Map()
/** @type {number} */
_redundantSnapshotCount = 0
/** @type {number} */
_lastCoreMarkAt = 0
/** @type {number|null} */
_idleWindowStart = null
/** @type {ReturnType<typeof setTimeout>|null} */
_idleWatchdogTimer = null
/** @type {number} Active lane item-processing depth (suppress idle-watchdog while > 0). */
_productiveBusyDepth = 0
/** @type {object[]} */
_diagnosticsLog = []
/** @type {Set<string>} Dedupe keys for deep reentrancy-origin warnings. */
_reentrancyOriginWarned = new Set()
/** @type {object} */
thresholds = {
default: {windowMs: 1000, maxFires: 40},
signals: {windowMs: 500, maxFires: 80, maxDepth: 4, writeStormPerMacrotask: 24},
scheduler: {windowMs: 1000, maxFires: 60, spawnWindowMs: 1000, maxSpawns: 16, maxSpawnDepth: 8},
kernel: {windowMs: 1000, maxFires: 40},
lane: {windowMs: 1000, maxRearm: 8, cooldownMs: 2000},
effect: {windowMs: 1000, maxFires: 30},
}
constructor()
{
this._refreshToggles()
this._replayLastIncident()
this._bindIncidentLifecycle()
this._scheduleIdleWatchdogCheck()
}
/**
* Opt-out only: `localStorage['brazen:reactor:profile'] = 'disabled'` turns hooks off
* for baseline comparison. Verbose only adds per-mark console.debug spam.
* Capture, breakers, and incident persistence are on by default.
*/
_refreshToggles()
{
try {
let raw = globalThis.localStorage?.getItem(BRAZEN_REACTOR_PROFILE_STORAGE_KEY)
if (raw === '0' || raw === 'off' || raw === 'disabled') {
this.disabled = true
this.verbose = false
return
}
if (raw === '1' || raw === 'on' || raw === 'verbose') {
this.disabled = false
this.verbose = true
return
}
if (raw === 'quiet') {
this.disabled = false
this.verbose = false
return
}
} catch (e) {
// ignore storage failures
}
this.disabled = false
this.verbose = false
}
/** @return {boolean} */
active()
{
return !this.disabled
}
/**
* @param {{kind: string, label?: string, detail?: unknown, parentId?: number|null}} cause
* @return {number}
*/
pushCause(cause)
{
if (this.disabled) {
return 0
}
let parentId = cause.parentId !== undefined
? cause.parentId
: (this._causeStack.length ? this._causeStack[this._causeStack.length - 1] : null)
let causeId = ++this._causeSeq
this._causes.set(causeId, {
causeId,
parentId,
kind: cause.kind,
label: cause.label ?? cause.kind,
detail: cause.detail ?? null,
rowCount: 0,
totalMs: 0,
reentrantCount: 0,
})
while (this._causes.size > BRAZEN_REACTOR_CAUSE_CAP) {
let oldest = this._causes.keys().next().value
this._causes.delete(oldest)
}
this._causeStack.push(causeId)
return causeId
}
/**
* @return {number|undefined}
*/
popCause()
{
return this._causeStack.pop()
}
/**
* @return {{causeId: number, parentId: number|null, kind: string, label: string, detail: unknown}|null}
*/
activeCause()
{
if (!this._causeStack.length) {
return null
}
return this._causes.get(this._causeStack[this._causeStack.length - 1]) ?? null
}
/**
* @return {number|null}
*/
activeCauseId()
{
return this._causeStack.length ? this._causeStack[this._causeStack.length - 1] : null
}
/**
* @return {string|null}
*/
activeCauseLabel()
{
let active = this.activeCause()
return active ? `${active.kind}:${active.label}` : null
}
/**
* @param {{kind: string, label?: string, detail?: unknown, parentId?: number|null}} cause
* @param {() => unknown} fn
* @return {unknown}
*/
withCause(cause, fn)
{
if (this.disabled) {
return fn()
}
this.noteReentrant(cause.kind)
this.pushCause(cause)
try {
let result = fn()
if (result && typeof result.then === 'function') {
return result.finally(() => {
this.popCause()
})
}
this.popCause()
return result
} catch (error) {
this.popCause()
throw error
}
}
/**
* @param {string} kind
* @return {boolean}
*/
noteReentrant(kind)
{
if (this.disabled) {
return false
}
let reentrant = this._causeStack.some((id) => this._causes.get(id)?.kind === kind)
if (!reentrant) {
return false
}
if (this._causeStack.length >= 2) {
this._captureReentrancyOrigin(kind)
}
this._reentrantCounts.set(kind, (this._reentrantCounts.get(kind) ?? 0) + 1)
let active = this.activeCause()
if (active) {
active.reentrantCount += 1
}
this.mark('cause', 'reentrant', {
key: kind,
data: {activeCause: this.activeCauseLabel()},
})
return true
}
/**
* Flattened cause stack for reentrancy diagnostics.
* @return {string}
* @private
*/
_formatCauseStack()
{
return this._causeStack.map((id) => {
let cause = this._causes.get(id)
return cause ? `${cause.kind}:${cause.label}` : '?'
}).join('<-')
}
/**
* Record and warn once when cause nesting exceeds depth 2 (loop ignition edge).
* @param {string} triggerKind
* @param {string|null|undefined} [triggerLabel]
* @private
*/
_captureReentrancyOrigin(triggerKind, triggerLabel = null)
{
if (this.disabled) {
return
}
let stack = this._formatCauseStack()
let label = triggerLabel ?? triggerKind
let tag = `${triggerKind}:${label}:${stack}`
if (this._reentrancyOriginWarned.has(tag)) {
return
}
this._reentrancyOriginWarned.add(tag)
while (this._reentrancyOriginWarned.size > 64) {
let oldest = this._reentrancyOriginWarned.values().next().value
this._reentrancyOriginWarned.delete(oldest)
}
let payload = {
depth: this._causeStack.length,
triggerKind,
triggerLabel: label,
stack,
activeCause: this.activeCauseLabel(),
}
this._recordDiagnostic('reentrancy', triggerKind, label, payload)
this._warnTextOnce(
`reentrancy:${triggerKind}:${stack}`,
`[BrazenReactor] reentrancy depth ${this._causeStack.length} at ${triggerKind}:${label} | stack=${stack} active=${payload.activeCause ?? '-'}`,
)
}
/**
* @param {string} type
* @param {number|null} [startCauseId]
* @return {boolean}
*/
causeChainHasJobType(type, startCauseId = null)
{
if (this.disabled || !type) {
return false
}
let id = startCauseId ?? this.activeCauseId()
while (id != null) {
let cause = this._causes.get(id)
if (!cause) {
break
}
if (cause.kind === 'job' && String(cause.label).startsWith(`${type}:`)) {
return true
}
id = cause.parentId
}
return false
}
/**
* @param {string} type
* @param {string|null} originCauseLabel
* @param {number|null} [originCauseId]
* @return {boolean}
*/
noteScheduledReentrant(type, originCauseLabel, originCauseId = null)
{
if (this.disabled || !type) {
return false
}
this._scheduledReentrantCounts.set(type, (this._scheduledReentrantCounts.get(type) ?? 0) + 1)
let samples = this._scheduledReentrantSamples.get(type)
if (!samples) {
samples = new Set()
this._scheduledReentrantSamples.set(type, samples)
}
if (originCauseLabel && samples.size < 8) {
samples.add(originCauseLabel)
}
this.mark('cause', 'scheduled-reentrant', {
key: type,
data: {originCause: originCauseLabel, originCauseId},
})
return true
}
/**
* @return {object[]}
*/
_buildCauseBreakdown(limit = 20)
{
let rows = [...this._causes.values()].map((cause) => ({
causeId: cause.causeId,
parentId: cause.parentId,
kind: cause.kind,
label: cause.label,
rowCount: cause.rowCount,
totalMs: Number(cause.totalMs.toFixed(2)),
reentrantCount: cause.reentrantCount,
}))
rows.sort((a, b) => b.rowCount - a.rowCount || b.totalMs - a.totalMs)
return rows.slice(0, limit)
}
/**
* @param {number} [limit]
* @return {{path: string, count: number}[]}
*/
_buildTopPaths(limit = 10)
{
return [...this._pathFanout.entries()]
.map(([path, count]) => ({path, count}))
.sort((a, b) => b.count - a.count)
.slice(0, limit)
}
/**
* @param {object} [payload]
* @return {string}
*/
_formatCauseSummary(payload = {})
{
/** @type {string[]} */
let lines = ['[BrazenReactor] incident summary']
if (payload.breakerId) {
lines.push(`breaker=${payload.breakerId}`)
}
if (payload.busyMs != null) {
lines.push(`busyMs=${payload.busyMs}`)
}
let topCauses = payload.topCauses ?? payload.causeBreakdown ?? this._buildCauseBreakdown(12)
if (topCauses?.length) {
lines.push('topCauses:')
for (let cause of topCauses.slice(0, 12)) {
lines.push(
` ${cause.kind}:${cause.label} parent=${cause.parentId ?? '-'} rows=${cause.rowCount} ms=${cause.totalMs} re=${cause.reentrantCount ?? 0}`,
)
}
}
let channels = payload.channelCounts ?? {}
let channelKeys = Object.keys(channels)
if (channelKeys.length) {
lines.push(`channels: ${channelKeys.map((key) => `${key}=${channels[key]}`).join(' ')}`)
}
let reentrant = payload.reentrant ?? Object.fromEntries(this._reentrantCounts)
if (Object.keys(reentrant).length) {
lines.push(`reentrant: ${JSON.stringify(reentrant)}`)
}
let scheduledReentrant = payload.scheduledReentrant ?? Object.fromEntries(this._scheduledReentrantCounts)
if (Object.keys(scheduledReentrant).length) {
lines.push(`scheduledReentrant: ${JSON.stringify(scheduledReentrant)}`)
let sampleParts = []
for (let [type, samples] of this._scheduledReentrantSamples) {
if (samples.size) {
sampleParts.push(`${type}=[${[...samples].join(', ')}]`)
}
}
if (sampleParts.length) {
lines.push(`scheduledReentrantSamples: ${sampleParts.join(' ')}`)
}
}
let topPaths = payload.topPaths ?? this._buildTopPaths(10)
if (topPaths?.length) {
lines.push('topPaths:')
for (let row of topPaths) {
lines.push(` ${row.path}=${row.count}`)
}
}
let redundantSnapshot = payload.redundantSnapshot ?? this._redundantSnapshotCount
if (redundantSnapshot) {
lines.push(`redundantSnapshot=${redundantSnapshot}`)
}
return lines.join('\n')
}
/**
* @param {object} [payload]
* @return {string}
*/
_formatLoopContext(payload = {})
{
/** @type {string[]} */
let parts = [`cause=${payload.activeCause ?? '-'}`]
if (payload.topCauses?.length) {
parts.push('causes=' + payload.topCauses
.map((cause) => `${cause.kind}:${cause.label}(r${cause.rowCount},m${cause.totalMs})`)
.join(','))
}
if (payload.topPaths?.length) {
parts.push('paths=' + payload.topPaths.map((row) => `${row.path}:${row.count}`).join(','))
}
return parts.join(' ')
}
/**
* @param {string} kind
* @param {string} channel
* @param {string|undefined} key
* @param {object} [data]
*/
_recordDiagnostic(kind, channel, key, data = {})
{
if (this.disabled) {
return
}
this._diagnosticsLog.push({
t: Date.now(),
kind,
channel,
key: key ?? undefined,
...data,
})
if (this._diagnosticsLog.length > BRAZEN_REACTOR_DIAG_CAP) {
this._diagnosticsLog.splice(0, this._diagnosticsLog.length - BRAZEN_REACTOR_DIAG_CAP)
}
}
/**
* @param {object} [payload]
* @return {string}
*/
summary(payload)
{
return this._formatCauseSummary(payload ?? {
causeBreakdown: this._buildCauseBreakdown(12),
reentrant: Object.fromEntries(this._reentrantCounts),
scheduledReentrant: Object.fromEntries(this._scheduledReentrantCounts),
topPaths: this._buildTopPaths(10),
redundantSnapshot: this._redundantSnapshotCount,
})
}
/**
* @private
* @param {string} channel
* @param {string} event
* @param {string|undefined} key
* @param {unknown} data
*/
_aggregateMarkPaths(channel, event, key, data)
{
if (!data || typeof data !== 'object') {
if (channel === 'signals' && (event === 'commit' || event === 'atom-write') && key) {
this._pathFanout.set(String(key), (this._pathFanout.get(String(key)) ?? 0) + 1)
}
return
}
/** @type {string[]} */
let paths = []
if (data.path) {
paths.push(String(data.path))
}
if (Array.isArray(data.paths)) {
paths.push(...data.paths.map((path) => String(path)))
}
if (Array.isArray(data.dirtyPaths)) {
paths.push(...data.dirtyPaths.map((path) => String(path)))
}
if (channel === 'signals' && (event === 'commit' || event === 'atom-write') && key) {
paths.push(String(key))
}
for (let path of paths) {
if (!path) {
continue
}
this._pathFanout.set(path, (this._pathFanout.get(path) ?? 0) + 1)
}
while (this._pathFanout.size > BRAZEN_REACTOR_PATH_FANOUT_CAP) {
let oldest = this._pathFanout.keys().next().value
this._pathFanout.delete(oldest)
}
if (data.redundantSnapshot === true) {
this._redundantSnapshotCount += 1
}
}
/**
* @private
*/
_noteCoreActivity(t)
{
if (this._idleWindowStart == null) {
this._idleWindowStart = t
}
this._lastCoreMarkAt = t
this._scheduleIdleWatchdogCheck()
}
/**
* @private
*/
_scheduleIdleWatchdogCheck()
{
if (this.disabled || this._idleWatchdogTimer != null) {
return
}
this._idleWatchdogTimer = setTimeout(() => {
this._idleWatchdogTimer = null
this._checkIdleWatchdog()
}, BRAZEN_REACTOR_IDLE_QUIET_MS)
}
/**
* @private
*/
_checkIdleWatchdog()
{
if (this.disabled) {
return
}
let now = Date.now()
let gap = now - this._lastCoreMarkAt
if (this._lastCoreMarkAt > 0 && gap >= BRAZEN_REACTOR_IDLE_QUIET_MS) {
this._idleWindowStart = null
return
}
if (this._idleWindowStart != null && (now - this._idleWindowStart) >= BRAZEN_REACTOR_IDLE_WINDOW_MS) {
if (this._productiveBusyDepth > 0) {
this._idleWindowStart = now
this._scheduleIdleWatchdogCheck()
return
}
let breakdown = this._buildCauseBreakdown(12)
let channels = {}
for (let row of this._ring.slice(-BRAZEN_REACTOR_INCIDENT_ROWS)) {
if (!BRAZEN_REACTOR_CORE_CHANNELS.has(row.channel)) {
continue
}
channels[row.channel] = (channels[row.channel] ?? 0) + 1
}
this._tripBreaker('idle-watchdog', '[BrazenReactor] never returned to idle', {
busyMs: now - this._idleWindowStart,
quietMs: BRAZEN_REACTOR_IDLE_QUIET_MS,
idleWindowMs: BRAZEN_REACTOR_IDLE_WINDOW_MS,
topCauses: breakdown,
channelCounts: channels,
reentrant: Object.fromEntries(this._reentrantCounts),
scheduledReentrant: Object.fromEntries(this._scheduledReentrantCounts),
topPaths: this._buildTopPaths(10),
redundantSnapshot: this._redundantSnapshotCount,
})
this._idleWindowStart = now
}
this._scheduleIdleWatchdogCheck()
}
/**
* @param {string} channel
* @param {string} event
* @param {{key?: string, data?: unknown, ms?: number}} [options]
*/
mark(channel, event, options = {})
{
if (this.disabled) {
return
}
let causeId = this.activeCauseId()
let row = {
t: Date.now(),
channel,
event,
key: options.key,
data: options.data,
ms: options.ms,
causeId: causeId ?? undefined,
}
this._pushRow(row)
this._trackFire(channel, options.key ?? event, row.t)
if (BRAZEN_REACTOR_CORE_CHANNELS.has(channel)) {
this._noteCoreActivity(row.t)
}
if (causeId != null) {
let cause = this._causes.get(causeId)
if (cause) {
cause.rowCount += 1
if (typeof options.ms === 'number') {
cause.totalMs += options.ms
}
}
}
this._aggregateMarkPaths(channel, event, options.key, options.data)
if (this.verbose) {
console.debug('[BrazenReactor]', channel, event, options.key ?? '', causeId ?? '', options.data ?? '')
}
}
/**
* @param {string} channel
* @param {string} label
* @param {{key?: string}} [options]
* @return {() => void}
*/
time(channel, label, options = {})
{
if (this.disabled) {
return () => {}
}
let start = performance.now()
return () => {
let ms = performance.now() - start
this.mark(channel, label, {key: options.key, ms})
if (ms >= BRAZEN_REACTOR_LONG_TASK_MS) {
this._recordDiagnostic('long-task', channel, label, {
ms: Number(ms.toFixed(1)),
key: options.key,
})
this._warnOnce(`${channel}:${label}:long-task`, `[BrazenReactor] long-task ${channel}/${label} ${ms.toFixed(1)}ms`, {
channel,
label,
key: options.key,
ms,
})
this._schedulePersistIncident(`long-task:${channel}/${label}`)
}
return ms
}
}
/**
* @param {string} channel
* @param {string} [key]
* @return {number}
*/
enter(channel, key = channel)
{
if (this.disabled) {
return 0
}
let depthKey = `${channel}\0${key}`
let next = (this._depth.get(depthKey) ?? 0) + 1
this._depth.set(depthKey, next)
this.mark(channel, 'enter', {key, data: {depth: next}})
return next
}
/**
* @param {string} channel
* @param {string} [key]
* @return {number}
*/
exit(channel, key = channel)
{
if (this.disabled) {
return 0
}
let depthKey = `${channel}\0${key}`
let next = Math.max(0, (this._depth.get(depthKey) ?? 1) - 1)
if (next === 0) {
this._depth.delete(depthKey)
} else {
this._depth.set(depthKey, next)
}
this.mark(channel, 'exit', {key, data: {depth: next}})
return next
}
/**
* @param {string} channel
* @param {string} [key]
* @return {number}
*/
depth(channel, key = channel)
{
return this._depth.get(`${channel}\0${key}`) ?? 0
}
/**
* @param {string} path
* @return {boolean} true when write should proceed; false when breaker skips redundant propagation
*/
noteAtomWrite(path)
{
if (this.disabled) {
return true
}
let now = Date.now()
if (this._writeStormMacrotask == null) {
this._writeStormMacrotask = now
this._writeStormCounts.clear()
queueMicrotask(() => {
this._writeStormMacrotask = null
this._writeStormCounts.clear()
})
}
let count = (this._writeStormCounts.get(path) ?? 0) + 1
this._writeStormCounts.set(path, count)
this.mark('signals', 'atom-write', {
key: path,
data: {count, cause: this.activeCauseLabel()},
})
let limit = this.thresholds.signals.writeStormPerMacrotask
if (count > limit) {
this._tripBreaker('signals:write-storm', `[BrazenReactor] write-storm on ${path} (${count} writes in macrotask)`, {
path,
count,
limit,
})
return false
}
return true
}
/**
* @param {string} parentKey
* @param {string} childType
* @return {boolean}
*/
noteSpawn(parentKey, childType)
{
if (this.disabled) {
return true
}
let key = `${parentKey}\0${childType}`
let t = Date.now()
this._trackFire('scheduler', key, t, this.thresholds.scheduler)
let depth = (this._spawnContext.get(parentKey)?.depth ?? 0) + 1
this._spawnContext.set(parentKey, {rootKey: parentKey, depth})
this.mark('scheduler', 'spawn', {key: parentKey, data: {childType, depth}})
if (depth > this.thresholds.scheduler.maxSpawnDepth) {
this._tripBreaker('scheduler:spawn-depth', `[BrazenReactor] spawn depth exceeded for ${parentKey}`, {
parentKey,
childType,
depth,
})
return false
}
if (this._isSuspectedLoop('scheduler', key)) {
this._tripBreaker('scheduler:spawn-storm', `[BrazenReactor] spawn storm ${childType} under ${parentKey}`, {
parentKey,
childType,
})
return false
}
return true
}
/**
* @param {string} laneId
* @param {'reentry'|'rearm'|'item'} kind
* @param {{itemsProcessed?: number}} [options]
* @return {boolean}
*/
noteLane(laneId, kind, options = {})
{
if (this.disabled) {
return true
}
let key = `${laneId}:${kind}`
this.mark('lane', kind, {key: laneId, data: options})
if (kind === 'rearm') {
this._trackFire('lane', `${laneId}:rearm`, Date.now(), this.thresholds.lane)
if (this._isSuspectedLoop('lane', `${laneId}:rearm`) &&
(options.itemsProcessed ?? 0) === 0) {
this._tripBreaker('lane:rearm-storm', `[BrazenReactor] lane re-arm storm on ${laneId}`, {
laneId,
itemsProcessed: options.itemsProcessed ?? 0,
})
let until = Date.now() + this.thresholds.lane.cooldownMs
this._cooldownUntil.set(`lane:${laneId}:rearm`, until)
return false
}
}
if (kind === 'reentry') {
this._trackFire('lane', `${laneId}:reentry`, Date.now(), this.thresholds.lane)
}
return !this._isLaneCooldown(laneId)
}
/** Lane processors are actively working (defer idle-watchdog incidents). */
enterProductiveBusy()
{
this._productiveBusyDepth += 1
}
/** @return {void} */
exitProductiveBusy()
{
this._productiveBusyDepth = Math.max(0, this._productiveBusyDepth - 1)
}
/**
* @param {string} laneId
* @return {boolean}
*/
_isLaneCooldown(laneId)
{
let until = this._cooldownUntil.get(`lane:${laneId}:rearm`) ?? 0
return until > Date.now()
}
/**
* @param {string} channel
* @param {string} key
* @return {boolean}
*/
shouldSkipPropagation(channel, key = 'propagate')
{
if (this.disabled) {
return false
}
let depth = this.depth(channel, key)
let maxDepth = this.thresholds.signals.maxDepth
if (depth >= maxDepth) {
this._tripBreaker('signals:reentrancy', `[BrazenReactor] propagation re-entrancy depth ${depth}`, {
depth,
maxDepth,
})
return true
}
return false
}
/**
* @param {string} channel
* @param {string} key
* @param {object} [override]
* @return {boolean}
*/
_isSuspectedLoop(channel, key, override = null)
{
let cfg = override ?? this.thresholds[channel] ?? this.thresholds.default
let times = this._fireTimes.get(`${channel}\0${key}`) ?? []
let windowMs = cfg.windowMs ?? this.thresholds.default.windowMs
let maxFires = cfg.maxFires ?? cfg.maxRearm ?? this.thresholds.default.maxFires
let cutoff = Date.now() - windowMs
times = times.filter((t) => t >= cutoff)
this._fireTimes.set(`${channel}\0${key}`, times)
if (times.length >= maxFires) {
let statKey = `${channel}\0${key}`
let stat = this._stats.get(statKey)
if (stat) {
stat.suspected = true
}
return true
}
return false
}
/**
* @param {string} channel
* @param {string} key
* @param {number} t
* @param {object} [override]
*/
_trackFire(channel, key, t, override = null)
{
let statKey = `${channel}\0${key}`
let stat = this._stats.get(statKey) ?? {
count: 0,
totalMs: 0,
maxMs: 0,
lastT: 0,
suspected: false,
}
stat.count += 1
stat.lastT = t
this._stats.set(statKey, stat)
let times = this._fireTimes.get(statKey) ?? []
times.push(t)
let cfg = override ?? this.thresholds[channel] ?? this.thresholds.default
let windowMs = cfg.windowMs ?? this.thresholds.default.windowMs
let cutoff = t - windowMs
while (times.length && times[0] < cutoff) {
times.shift()
}
this._fireTimes.set(statKey, times)
let maxFires = cfg.maxFires ?? cfg.maxRearm ?? this.thresholds.default.maxFires
if (times.length >= maxFires) {
stat.suspected = true
let payload = {
activeCause: this.activeCauseLabel(),
topCauses: this._buildCauseBreakdown(6),
topPaths: this._buildTopPaths(6),
}
this._recordDiagnostic('loop', channel, key, {
count: times.length,
windowMs,
...payload,
})
this._warnTextOnce(
`${channel}:${key}:loop`,
`[BrazenReactor] suspected loop ${channel}/${key} (${times.length} in ${windowMs}ms) | ${this._formatLoopContext(payload)}`,
)
}
}
/**
* @param {BrazenReactorProfileRow} row
*/
_pushRow(row)
{
this._ring.push(row)
if (this._ring.length > BRAZEN_REACTOR_RING_CAP) {
this._ring.splice(0, this._ring.length - BRAZEN_REACTOR_RING_CAP)
}
if (typeof row.ms === 'number') {
let statKey = `${row.channel}\0${row.key ?? row.event}`
let stat = this._stats.get(statKey) ?? {
count: 0,
totalMs: 0,
maxMs: 0,
lastT: 0,
suspected: false,
}
stat.count += 1
stat.totalMs += row.ms
stat.maxMs = Math.max(stat.maxMs, row.ms)
stat.lastT = row.t
this._stats.set(statKey, stat)
}
}
/**
* @param {string} id
* @param {string} message
* @param {object} [data]
*/
_warnOnce(id, message, data = {})
{
let now = Date.now()
let last = this._lastWarnAt.get(id) ?? 0
if (now - last < BRAZEN_REACTOR_WARN_COOLDOWN_MS) {
return
}
this._lastWarnAt.set(id, now)
if (data && Object.keys(data).length) {
console.warn(message, data)
} else {
console.warn(message)
}
}
/**
* @param {string} id
* @param {string} text
*/
_warnTextOnce(id, text)
{
let now = Date.now()
let last = this._lastWarnAt.get(id) ?? 0
if (now - last < BRAZEN_REACTOR_WARN_COOLDOWN_MS) {
return
}
this._lastWarnAt.set(id, now)
console.warn(text)
}
/**
* @param {string} id
* @param {string} message
* @param {object} [data]
*/
_tripBreaker(id, message, data = {})
{
this._breakerCounts.set(id, (this._breakerCounts.get(id) ?? 0) + 1)
let payload = {...data, breakers: this._breakerCounts.get(id), breakerId: id}
this._recordDiagnostic('incident', id, undefined, payload)
this._warnOnce(id, message, payload)
if (id === 'idle-watchdog' || id === 'signals:write-storm' || id === 'signals:reentrancy') {
this._warnTextOnce(`${id}:summary`, this._formatCauseSummary(payload))
}
this._surfaceIncident(message, payload)
this._schedulePersistIncident(id)
}
/**
* @param {string} reason
* @private
*/
_schedulePersistIncident(reason)
{
if (this.disabled) {
return
}
if (this._persistTimer != null) {
clearTimeout(this._persistTimer)
}
this._persistTimer = setTimeout(() => {
this._persistTimer = null
this._persistIncident(reason)
}, BRAZEN_REACTOR_PERSIST_DEBOUNCE_MS)
}
/**
* @param {string} reason
* @private
*/
_persistIncident(reason)
{
if (this.disabled) {
return
}
let incident = this._buildIncidentSnapshot(reason)
this.lastIncident = incident
try {
globalThis.sessionStorage?.setItem(BRAZEN_REACTOR_INCIDENT_STORAGE_KEY, JSON.stringify(incident))
} catch (e) {
// ignore quota / privacy mode
}
}
/**
* @param {string} reason
* @return {object}
* @private
*/
_buildIncidentSnapshot(reason)
{
let table = []
for (let [statKey, stat] of this._stats) {
if (!stat.suspected && stat.maxMs < BRAZEN_REACTOR_LONG_TASK_MS) {
continue
}
let split = statKey.indexOf('\0')
table.push({
channel: split >= 0 ? statKey.slice(0, split) : statKey,
key: split >= 0 ? statKey.slice(split + 1) : '',
count: stat.count,
maxMs: Number(stat.maxMs.toFixed(2)),
suspected: stat.suspected,
})
}
table.sort((a, b) => b.maxMs - a.maxMs || b.count - a.count)
return {
savedAt: Date.now(),
reason,
breakers: Object.fromEntries(this._breakerCounts),
topStats: table.slice(0, 20),
causeBreakdown: this._buildCauseBreakdown(20),
reentrant: Object.fromEntries(this._reentrantCounts),
scheduledReentrant: Object.fromEntries(this._scheduledReentrantCounts),
topPaths: this._buildTopPaths(10),
redundantSnapshot: this._redundantSnapshotCount,
summary: this.summary(),
timeline: this._ring.slice(-BRAZEN_REACTOR_INCIDENT_ROWS),
depth: Object.fromEntries(this._depth),
}
}
/**
* @private
*/
_replayLastIncident()
{
try {
let raw = globalThis.sessionStorage?.getItem(BRAZEN_REACTOR_INCIDENT_STORAGE_KEY)
if (!raw) {
return
}
let incident = JSON.parse(raw)
this.lastIncident = incident
let ageSec = Math.round((Date.now() - (incident.savedAt ?? 0)) / 1000)
console.warn(
`[BrazenReactor] previous incident (${ageSec}s ago): ${incident.reason ?? 'unknown'}`,
incident.breakers ?? {},
'Full snapshot: globalThis.__brazenReactor.lastIncident',
)
} catch (e) {
// ignore corrupt snapshot
}
}
/**
* @param {string} message
* @param {object} data
* @private
*/
_surfaceIncident(message, data)
{
if (typeof document === 'undefined') {
return
}
let root = document.getElementById('brazen-reactor-incident')
if (!root) {
root = document.createElement('div')
root.id = 'brazen-reactor-incident'
root.style.cssText = [
'position:fixed',
'right:8px',
'bottom:8px',
'z-index:2147483646',
'max-width:min(420px,90vw)',
'padding:8px 10px',
'border-radius:6px',
'background:rgba(20,20,24,.92)',
'color:#f5f5f5',
'font:12px/1.35 ui-monospace,monospace',
'box-shadow:0 4px 18px rgba(0,0,0,.35)',
'pointer-events:auto',
].join(';')
document.documentElement.appendChild(root)
}
root.textContent = `${message} · breakers=${JSON.stringify(data.breakers ?? {})}`
}
/**
* @private
*/
_bindIncidentLifecycle()
{
if (this._lifecycleBound || typeof document === 'undefined') {
return
}
this._lifecycleBound = true
document.addEventListener('pagehide', () => {
if (this._breakerCounts.size > 0) {
this._persistIncident('pagehide')
}
})
document.addEventListener('visibilitychange', () => {
if (document.visibilityState === 'hidden' && this._breakerCounts.size > 0) {
this._persistIncident('tab-hidden')
}
})
}
clear()
{
this._ring = []
this._stats.clear()
this._fireTimes.clear()
this._depth.clear()
this._breakerCounts.clear()
this._cooldownUntil.clear()
this._writeStormCounts.clear()
this._spawnContext.clear()
this._causeStack = []
this._causes.clear()
this._reentrantCounts.clear()
this._scheduledReentrantCounts.clear()
this._scheduledReentrantSamples.clear()
this._pathFanout.clear()
this._redundantSnapshotCount = 0
this._lastCoreMarkAt = 0
this._idleWindowStart = null
this._diagnosticsLog = []
this._reentrancyOriginWarned.clear()
}
/**
* @return {string}
*/
exportLog()
{
return JSON.stringify({
exportedAt: Date.now(),
url: typeof location !== 'undefined' ? location.href : '',
diagnostics: this._diagnosticsLog,
summary: this.summary(),
}, null, 2)
}
clearLog()
{
this._diagnosticsLog = []
}
dump()
{
return {
rows: this._ring.slice(),
stats: Object.fromEntries(this._stats),
breakers: Object.fromEntries(this._breakerCounts),
depth: Object.fromEntries(this._depth),
cooldownUntil: Object.fromEntries(this._cooldownUntil),
causes: Object.fromEntries(this._causes),
causeBreakdown: this._buildCauseBreakdown(),
reentrant: Object.fromEntries(this._reentrantCounts),
scheduledReentrant: Object.fromEntries(this._scheduledReentrantCounts),
topPaths: this._buildTopPaths(10),
redundantSnapshot: this._redundantSnapshotCount,
summary: this.summary(),
activeCause: this.activeCause(),
}
}
report()
{
let table = []
for (let [statKey, stat] of this._stats) {
let split = statKey.indexOf('\0')
table.push({
channel: split >= 0 ? statKey.slice(0, split) : statKey,
key: split >= 0 ? statKey.slice(split + 1) : '',
count: stat.count,
totalMs: Number(stat.totalMs.toFixed(2)),
maxMs: Number(stat.maxMs.toFixed(2)),
lastT: stat.lastT,
suspected: stat.suspected,
})
}
table.sort((a, b) => b.count - a.count)
let causeBreakdown = this._buildCauseBreakdown(20)
console.group('[BrazenReactor] report')
console.table(table.slice(0, 40))
console.table(causeBreakdown)
console.log('breakers', Object.fromEntries(this._breakerCounts))
console.log('reentrant', Object.fromEntries(this._reentrantCounts))
console.log('scheduledReentrant', Object.fromEntries(this._scheduledReentrantCounts))
console.log('topPaths', this._buildTopPaths(10))
console.log('redundantSnapshot', this._redundantSnapshotCount)
console.log(this.summary())
console.log('activeCause', this.activeCause())
console.log('recent timeline', this._ring.slice(-40))
console.groupEnd()
return this.dump()
}
}
/** @type {() => BrazenReactorProfiler} */
function brazenReactorProfiler()
{
return BrazenReactorProfiler.get()
}
globalThis.BrazenReactorProfiler = BrazenReactorProfiler
globalThis.__brazenReactor = brazenReactorProfiler()
// -------------------------------------------------------------------------
// Signals
// -------------------------------------------------------------------------
/**
* @typedef {() => void} Unsubscribe
*/
/**
* @typedef {Object} Atom
* @property {unknown} value
* @property {number} version
* @property {(next: unknown) => void} set
* @property {(fn: (value: unknown) => void) => Unsubscribe} subscribe
*/
/**
* @typedef {Object} Computed
* @property {unknown} value
* @property {number} version
* @property {(fn: (value: unknown) => void) => Unsubscribe} subscribe
*/
/**
* @typedef {Object} Effect
* @property {() => void} dispose
*/
class BrazenSignalCycleError extends Error
{
constructor(message = 'Signal dependency cycle detected')
{
super(message)
this.name = 'BrazenSignalCycleError'
}
}
/** @type {number} */
let propagationGeneration = 0
/** @type {number} */
let batchDepth = 0
/** @type {Set<InternalNode>} */
let pendingDirtyAtoms = new Set()
/**
* @typedef {Object} TrackingFrame
* @property {InternalConsumer|null} owner
* @property {Set<InternalNode>} deps
* @property {InternalComputed[]} stack
* @property {boolean} [forceFresh]
*/
/** @type {Set<InternalComputed>} */
const initializingComputeds = new Set()
/** @type {TrackingFrame|null} */
let trackingFrame = null
/**
* @typedef {Object} InternalNode
* @property {'atom'|'computed'} kind
* @property {Set<InternalConsumer>} dependents
*/
/**
* @typedef {Object} InternalConsumer
* @property {'computed'|'effect'} kind
* @property {Map<InternalNode, number>} depVersions
* @property {Set<InternalNode>} depNodes
* @property {Set<(value: unknown) => void>} subscribers
*/
/**
* @param {InternalConsumer} consumer
* @param {InternalNode} producer
*/
function linkDependency(consumer, producer)
{
if (!consumer.depVersions.has(producer)) {
consumer.depNodes.add(producer)
consumer.depVersions.set(producer, producer.version)
producer.dependents.add(consumer)
}
else if (trackingFrame) {
consumer.depVersions.set(producer, producer.version)
}
}
/**
* @param {InternalConsumer} consumer
*/
function clearConsumerDependencies(consumer)
{
for (const producer of consumer.depNodes) {
producer.dependents.delete(consumer)
}
consumer.depNodes.clear()
consumer.depVersions.clear()
}
/**
* @param {InternalComputed} computed
* @returns {boolean}
*/
function isComputedStale(computed)
{
for (const [dep, seenVersion] of computed.depVersions) {
if (dep.version !== seenVersion) {
return true
}
}
return false
}
/**
* @param {InternalComputed} computed
*/
function assertNoCycle(computed)
{
if (trackingFrame && trackingFrame.stack.includes(computed)) {
throw new BrazenSignalCycleError('Signal dependency cycle detected')
}
}
/**
* @typedef {InternalConsumer & {
* kind: 'computed',
* read: () => unknown,
* cachedValue: unknown,
* version: number,
* path?: string,
* }} InternalAtom
*/
/**
* @typedef {InternalNode & {
* kind: 'atom',
* path?: string,
* cachedValue: unknown,
* version: number,
* set: (next: unknown) => void,
* get value: () => unknown,
* subscribe: (fn: (value: unknown) => void) => Unsubscribe,
* }} InternalAtom
*/
/**
* @typedef {InternalConsumer & {
* kind: 'computed',
* read: () => unknown,
* cachedValue: unknown,
* version: number,
* name?: string,
* get value: () => unknown,
* subscribe: (fn: (value: unknown) => void) => Unsubscribe,
* }} InternalComputed
*/
/**
* @typedef {InternalConsumer & {
* kind: 'effect',
* run: () => void,
* disposed: boolean,
* cleanup: (() => void)|null,
* name?: string,
* }} InternalEffect
*/
/**
* @param {InternalComputed} computed
* @returns {unknown}
*/
function recomputeComputed(computed)
{
assertNoCycle(computed)
const previousFrame = trackingFrame
const frame = {
owner: computed,
deps: new Set(),
stack: previousFrame ? [...previousFrame.stack, computed] : [computed],
forceFresh: previousFrame?.forceFresh,
}
trackingFrame = frame
clearConsumerDependencies(computed)
let nextValue
try {
nextValue = computed.read()
}
finally {
trackingFrame = previousFrame
}
for (const dep of frame.deps) {
linkDependency(computed, dep)
}
computed.cachedValue = nextValue
computed.version = computed.depNodes.size === 0
? propagationGeneration
: [...computed.depNodes].reduce((max, dep) => Math.max(max, dep.version), 0)
return nextValue
}
/**
* @param {InternalComputed} computed
* @returns {unknown}
*/
function readComputedValue(computed)
{
assertNoCycle(computed)
if (trackingFrame?.owner && trackingFrame.owner !== computed) {
trackingFrame.deps.add(computed)
linkDependency(trackingFrame.owner, computed)
}
if (isComputedStale(computed) || trackingFrame?.forceFresh) {
recomputeComputed(computed)
}
return computed.cachedValue
}
/**
* @param {InternalAtom} atom
* @returns {unknown}
*/
function readAtomValue(atom)
{
if (trackingFrame && trackingFrame.owner) {
trackingFrame.deps.add(atom)
linkDependency(trackingFrame.owner, atom)
}
return atom.cachedValue
}
/**
* @param {Set<InternalAtom>} dirtyAtoms
*/
function propagateFromDirtyAtoms(dirtyAtoms)
{
if (dirtyAtoms.size === 0) {
return
}
let profiler = brazenReactorProfiler()
if (profiler.shouldSkipPropagation('signals', 'propagate')) {
return
}
let depth = profiler.enter('signals', 'propagate')
let stop = profiler.time('signals', 'propagate', {key: `depth:${depth}`})
propagationGeneration++
/** @type {Set<InternalComputed>} */
const affectedComputeds = new Set()
/** @type {InternalNode[]} */
const queue = [...dirtyAtoms]
while (queue.length > 0) {
const node = queue.pop()
for (const dependent of node.dependents) {
if (dependent.kind === 'computed') {
if (!affectedComputeds.has(dependent)) {
affectedComputeds.add(dependent)
queue.push(dependent)
}
}
}
}
const sortedComputeds = topoSortComputeds(affectedComputeds)
/** @type {Map<InternalConsumer, unknown>} */
const previousValues = new Map()
for (const computed of sortedComputeds) {
previousValues.set(computed, computed.cachedValue)
recomputeComputed(computed)
}
const affectedEffects = collectTransitiveEffects([...dirtyAtoms, ...sortedComputeds])
let effectsRun = 0
for (const effect of affectedEffects) {
if (effectNeedsRun(effect)) {
runEffect(effect)
effectsRun += 1
}
}
for (const atom of dirtyAtoms) {
notifySubscribers(atom, atom.cachedValue)
}
for (const computed of sortedComputeds) {
if (!Object.is(previousValues.get(computed), computed.cachedValue)) {
notifySubscribers(computed, computed.cachedValue)
}
}
let ms = stop()
let dirtyPaths = [...dirtyAtoms].map((atom) => atom.path ?? 'atom').slice(0, 32)
profiler.mark('signals', 'propagate-end', {
key: `depth:${depth}`,
data: {
dirtyAtoms: dirtyAtoms.size,
dirtyPaths,
computeds: sortedComputeds.length,
effectsRun,
ms,
cause: profiler.activeCauseLabel(),
},
})
profiler.exit('signals', 'propagate')
}
/**
* @param {InternalNode[]} roots
* @returns {Set<InternalEffect>}
*/
function collectTransitiveEffects(roots)
{
/** @type {Set<InternalEffect>} */
const effects = new Set()
/** @type {InternalNode[]} */
const queue = [...roots]
/** @type {Set<InternalNode>} */
const visited = new Set()
while (queue.length > 0) {
const node = queue.pop()
if (!node || visited.has(node)) {
continue
}
visited.add(node)
for (const dependent of node.dependents) {
if (dependent.kind === 'effect' && !dependent.disposed) {
effects.add(dependent)
}
else if (dependent.kind === 'computed') {
queue.push(dependent)
}
}
}
return effects
}
/**
* @param {Set<InternalComputed>} computeds
* @returns {InternalComputed[]}
*/
function topoSortComputeds(computeds)
{
/** @type {InternalComputed[]} */
const sorted = []
/** @type {Set<InternalComputed>} */
const visited = new Set()
/** @type {Set<InternalComputed>} */
const visiting = new Set()
/**
* @param {InternalComputed} computed
*/
function visit(computed)
{
if (visited.has(computed) || !computeds.has(computed)) {
return
}
if (visiting.has(computed)) {
throw new BrazenSignalCycleError('Signal dependency cycle detected')
}
visiting.add(computed)
for (const dep of computed.depNodes) {
if (dep.kind === 'computed') {
visit(dep)
}
}
visiting.delete(computed)
visited.add(computed)
sorted.push(computed)
}
for (const computed of computeds) {
visit(computed)
}
return sorted
}
/**
* @param {InternalEffect} effect
* @returns {boolean}
*/
function effectNeedsRun(effect)
{
for (const [dep, seenVersion] of effect.depVersions) {
if (dep.version !== seenVersion) {
return true
}
}
return false
}
/**
* @param {InternalEffect} effect
*/
function runEffect(effect)
{
if (effect.disposed) {
return
}
let profiler = brazenReactorProfiler()
let effectName = effect.name ?? 'effect'
profiler.mark('effect', 'run', {key: effectName})
if (effect.cleanup) {
effect.cleanup()
effect.cleanup = null
}
const previousFrame = trackingFrame
const frame = {
owner: effect,
deps: new Set(),
stack: previousFrame ? [...previousFrame.stack] : [],
forceFresh: previousFrame?.forceFresh,
}
trackingFrame = frame
clearConsumerDependencies(effect)
let stop = profiler.time('effect', 'run', {key: effectName})
try {
const result = effect.run()
if (typeof result === 'function') {
effect.cleanup = result
}
}
finally {
stop()
trackingFrame = previousFrame
}
for (const dep of frame.deps) {
linkDependency(effect, dep)
}
}
/**
* @param {InternalConsumer} node
* @param {unknown} value
*/
function notifySubscribers(node, value)
{
for (const fn of node.subscribers) {
fn(value)
}
}
/**
* @param {InternalAtom} atom
*/
function markAtomDirty(atom)
{
let profiler = brazenReactorProfiler()
let path = atom.path ?? 'atom'
if (!profiler.noteAtomWrite(path)) {
return
}
if (batchDepth > 0) {
pendingDirtyAtoms.add(atom)
return
}
propagateFromDirtyAtoms(new Set([atom]))
}
/** @type {Map<string, InternalAtom>} */
const pathRegistry = new Map()
/**
* @param {InternalAtom} atom
* @param {unknown | ((prev: unknown) => unknown)} next
* @return {number}
*/
function commitAtomWrite(atom, next)
{
let resolved = typeof next === 'function' ? next(atom.cachedValue) : next
if (Object.is(atom.cachedValue, resolved)) {
return atom.version
}
atom.cachedValue = resolved
atom.version++
brazenReactorProfiler().mark('signals', 'commit', {
key: atom.path ?? 'atom',
data: {cause: brazenReactorProfiler().activeCauseLabel()},
})
markAtomDirty(atom)
return atom.version
}
/**
* @param {string} path
* @param {unknown} initial
* @return {InternalAtom}
*/
function getOrCreatePathAtom(path, initial)
{
let existing = pathRegistry.get(path)
if (existing) {
return existing
}
/** @type {InternalAtom} */
let atom = {
kind: 'atom',
path,
cachedValue: initial,
version: 0,
dependents: new Set(),
subscribers: new Set(),
set(next) {
commitAtomWrite(atom, next)
},
get value() {
return readAtomValue(atom)
},
subscribe(fn) {
atom.subscribers.add(fn)
return () => {
atom.subscribers.delete(fn)
}
},
}
pathRegistry.set(path, atom)
return atom
}
/**
* @template T
* @param {T} initial
* @param {{ path?: string }} [options]
* @returns {Atom<T>}
*/
function createAtom(initial, options = {})
{
if (options.path) {
return getOrCreatePathAtom(options.path, initial)
}
/** @type {InternalAtom} */
const atom = {
kind: 'atom',
path: options.path,
cachedValue: initial,
version: 0,
dependents: new Set(),
subscribers: new Set(),
set(next) {
commitAtomWrite(atom, next)
},
get value() {
return readAtomValue(atom)
},
subscribe(fn) {
atom.subscribers.add(fn)
return () => {
atom.subscribers.delete(fn)
}
},
}
return atom
}
/**
* @template T
* @param {() => T} read
* @param {{ name?: string }} [options]
* @returns {Computed<T>}
*/
function createComputed(read, options = {})
{
/** @type {InternalComputed} */
const computed = {
kind: 'computed',
name: options.name,
read,
cachedValue: undefined,
version: 0,
dependents: new Set(),
depNodes: new Set(),
depVersions: new Map(),
subscribers: new Set(),
get value() {
return readComputedValue(computed)
},
subscribe(fn) {
computed.subscribers.add(fn)
return () => {
computed.subscribers.delete(fn)
}
},
}
initializingComputeds.add(computed)
try {
recomputeComputed(computed)
}
finally {
initializingComputeds.delete(computed)
}
return computed
}
/**
* @param {() => void | (() => void)} fn
* @param {{ name?: string }} [options]
* @returns {Effect}
*/
function createEffect(fn, options = {})
{
/** @type {InternalEffect} */
const effect = {
kind: 'effect',
name: options.name,
run: fn,
disposed: false,
cleanup: null,
dependents: new Set(),
depNodes: new Set(),
depVersions: new Map(),
subscribers: new Set(),
}
runEffect(effect)
return {
dispose() {
if (effect.disposed) {
return
}
effect.disposed = true
if (effect.cleanup) {
effect.cleanup()
effect.cleanup = null
}
clearConsumerDependencies(effect)
},
}
}
/**
* @param {() => void} fn
*/
function batch(fn)
{
batchDepth++
let batchSize = pendingDirtyAtoms.size
try {
fn()
}
finally {
batchDepth--
if (batchDepth === 0 && pendingDirtyAtoms.size > 0) {
const dirty = pendingDirtyAtoms
pendingDirtyAtoms = new Set()
brazenReactorProfiler().mark('signals', 'batch-flush', {
data: {pendingBefore: batchSize, dirty: dirty.size},
})
propagateFromDirtyAtoms(dirty)
}
}
}
/**
* @returns {number}
*/
function getPropagationGeneration()
{
return propagationGeneration
}
/**
* @param {() => unknown} [read]
*/
function assertAcyclic(read)
{
if (typeof read !== 'function') {
return
}
const previousFrame = trackingFrame
trackingFrame = {
owner: null,
deps: new Set(),
stack: [],
forceFresh: true,
}
try {
read()
}
finally {
trackingFrame = previousFrame
}
}
/**
* @template T
* @param {string} path
* @param {T} initial
*/
function atom(path, initial)
{
let internal = getOrCreatePathAtom(path, initial)
return {
path,
read: () => readAtomValue(internal),
write: (next) => commitAtomWrite(internal, next),
peekVersion: () => internal.version,
get value() {
return readAtomValue(internal)
},
get version() {
return internal.version
},
set(next) {
commitAtomWrite(internal, next)
},
subscribe(fn) {
return internal.subscribe(fn)
},
}
}
function replicaAtom(path, initial)
{
return atom(path, initial)
}
function computed(path, deps, fn)
{
let internal = createComputed(() => {
if (typeof deps === 'function') {
deps()
}
return fn()
}, {name: path})
return {
path,
read: () => readComputedValue(internal),
peekVersion: () => internal.version,
subscribe(fn) {
return internal.subscribe(fn)
},
}
}
/**
* @param {() => void} deps
* @param {() => void | (() => void)} fn
* @param {string} [name]
*/
function effect(deps, fn, name)
{
return createEffect(() => {
if (typeof deps === 'function') {
deps()
}
return fn()
}, {name: name ?? 'effect'})
}
function applyPatches(patches)
{
if (!Array.isArray(patches)) {
return
}
batch(() => {
for (const patch of patches) {
if (!patch?.path || typeof patch.version !== 'number') {
continue
}
let internal = pathRegistry.get(patch.path)
if (!internal) {
internal = getOrCreatePathAtom(patch.path, patch.value)
}
if (patch.version <= internal.version) {
continue
}
internal.cachedValue = patch.value
internal.version = patch.version
markAtomDirty(internal)
}
})
}
function markDependency(target)
{
if (!trackingFrame?.owner || !target) {
return
}
if (typeof target.read === 'function') {
target.read()
} else if ('value' in target) {
void target.value
}
}
const BrazenSignals = Object.freeze({
atom,
replicaAtom,
computed,
effect,
batch,
applyPatches,
markDependency,
assertAcyclic,
BrazenSignalCycleError,
createAtom,
createComputed,
createEffect,
getPropagationGeneration,
})
globalThis.BrazenSignals = BrazenSignals
globalThis.createAtom = createAtom
globalThis.createComputed = createComputed
globalThis.createEffect = createEffect
globalThis.batch = batch
globalThis.getPropagationGeneration = getPropagationGeneration
globalThis.assertAcyclic = assertAcyclic
globalThis.BrazenSignalCycleError = BrazenSignalCycleError
// -------------------------------------------------------------------------
// EventBus
// -------------------------------------------------------------------------
/** @typedef {string} TabId */
/** @typedef {() => void} Unsubscribe */
/**
* @typedef {object} SignalPatch
* @property {string} path
* @property {number} version
* @property {unknown} value
*/
/**
* @typedef {object} BusMessageBase
* @property {'command' | 'patch' | 'snapshot-request' | 'snapshot-response'} kind
* @property {TabId} tabId
* @property {number} [seq]
* @property {number} [ts]
*/
/**
* @typedef {object} CommandMessage
* @property {'command'} kind
* @property {TabId} tabId
* @property {object} command
*/
/**
* @typedef {object} PatchMessage
* @property {'patch'} kind
* @property {TabId} tabId
* @property {number} seq
* @property {SignalPatch[]} patches
* @property {object[]} [events]
*/
/**
* @typedef {object} SnapshotRequestMessage
* @property {'snapshot-request'} kind
* @property {TabId} tabId
* @property {string} requestId
* @property {number} sinceSeq
*/
/**
* @typedef {object} SnapshotResponseMessage
* @property {'snapshot-response'} kind
* @property {TabId} tabId
* @property {string} requestId
* @property {Record<string, unknown>} snapshot
* @property {number} snapshotSeq
* @property {SignalPatch[]} [catchUpPatches]
*/
/** @typedef {CommandMessage | PatchMessage | SnapshotRequestMessage | SnapshotResponseMessage} BusMessage */
/**
* @typedef {object} SnapshotResponderResult
* @property {Record<string, unknown>} snapshot
* @property {number} snapshotSeq
* @property {SignalPatch[]} [catchUpPatches]
*/
/**
* @callback SnapshotResponder
* @param {SnapshotRequestMessage} request
* @return {SnapshotResponderResult | Promise<SnapshotResponderResult>}
*/
/**
* @callback BroadcastChannelFactory
* @param {string} channelName
* @return {{ postMessage: (data: unknown) => void, close: () => void, onmessage: ((event: { data: unknown }) => void) | null }}
*/
/**
* @typedef {object} BrazenEventBusOptions
* @property {TabId} [tabId]
* @property {SnapshotResponder} [onSnapshotRequest]
* @property {number} [snapshotTimeoutMs]
* @property {BroadcastChannelFactory} [createBroadcastChannel]
*/
const DEFAULT_SNAPSHOT_TIMEOUT_MS = 5000
const MAX_DELIVERY_QUEUE = 4096
const GAP_RESYNC_WAIT_MS = 250
class BrazenEventBus
{
// -------------------------------------------------------------------------
// Static public methods
// -------------------------------------------------------------------------
/**
* @param {string} scriptPrefix
* @param {BrazenEventBusOptions} [options]
* @return {BrazenEventBus}
*/
static create(scriptPrefix, options = {})
{
return new BrazenEventBus(scriptPrefix, options)
}
// -------------------------------------------------------------------------
// Public instance fields
// -------------------------------------------------------------------------
/** @type {TabId} */
tabId
/** @type {string} */
channelName
// -------------------------------------------------------------------------
// Protected class variables
// -------------------------------------------------------------------------
/** @type {ReturnType<BroadcastChannelFactory> | null} */
_channel = null
/** @type {Set<(message: BusMessage) => void>} */
_localListeners = new Set()
/** @type {Set<(message: BusMessage) => void>} */
_subscribers = new Set()
/** @type {number} */
_seq = 0
/** @type {boolean} */
_disposed = false
/** @type {boolean} */
_snapshotReady = false
/** @type {ReturnType<typeof setTimeout>|null} */
_gapResyncTimer = null
/** @type {boolean} */
_gapResyncInFlight = false
/** @type {number} */
_nextExpectedPatchSeq = 0
/** @type {PatchMessage[]} */
_patchBuffer = []
/** @type {BusMessage[]} */
_deliveryQueue = []
/** @type {boolean} */
_delivering = false
/**
* @type {Map<string, { resolve: (message: SnapshotResponseMessage) => void, reject: (error: Error) => void, timer: ReturnType<typeof setTimeout> }>}
*/
_pendingSnapshots = new Map()
/** @type {SnapshotResponder | null} */
_snapshotResponder = null
/** @type {number} */
_snapshotTimeoutMs = DEFAULT_SNAPSHOT_TIMEOUT_MS
/** @type {BroadcastChannelFactory} */
_createBroadcastChannel
// -------------------------------------------------------------------------
// Constructor
// -------------------------------------------------------------------------
/**
* @param {string} scriptPrefix
* @param {BrazenEventBusOptions} [options]
*/
constructor(scriptPrefix, options = {})
{
this.tabId = options.tabId ?? crypto.randomUUID()
this.channelName = `brazen-${scriptPrefix}`
this._snapshotResponder = options.onSnapshotRequest ?? null
this._snapshotTimeoutMs = options.snapshotTimeoutMs ?? DEFAULT_SNAPSHOT_TIMEOUT_MS
this._createBroadcastChannel = options.createBroadcastChannel ?? ((channelName) => new BroadcastChannel(channelName))
if (this._snapshotResponder) {
this._snapshotReady = true
}
this._openChannel()
}
// -------------------------------------------------------------------------
// Public instance methods
// -------------------------------------------------------------------------
/**
* In-tab + cross-tab publish. Followers: command only. Coordinator: all kinds.
* @param {BusMessage} message
*/
publish(message)
{
this._assertActive()
const stamped = this._stamp(message)
this._emitLocal(stamped)
this._postToChannel(stamped)
}
/**
* In-tab only publish (does not cross BroadcastChannel).
* @param {BusMessage} message
*/
emitLocal(message)
{
this._assertActive()
this._emitLocal(this._stamp(message))
}
/**
* Subscribe to in-tab events from publish/emitLocal only.
* @param {(message: BusMessage) => void} handler
* @return {Unsubscribe}
*/
onLocal(handler)
{
this._assertActive()
this._localListeners.add(handler)
return () => {
this._localListeners.delete(handler)
}
}
/**
* Ordered delivery per tab: commands FIFO; patches in seq order.
* @param {(message: BusMessage) => void} handler
* @return {Unsubscribe}
*/
subscribe(handler)
{
this._assertActive()
this._subscribers.add(handler)
return () => {
this._subscribers.delete(handler)
}
}
/**
* Coordinator: assign next seq (strictly monotonic with overflow guard).
* @return {number}
*/
nextSeq()
{
this._assertActive()
if (this._seq >= Number.MAX_SAFE_INTEGER) {
this._seq = 0
this._nextExpectedPatchSeq = 0
this._deliveryQueue = []
this._patchBuffer = []
return 1
}
this._seq += 1
return this._seq
}
/**
* Follower cold start: request full snapshot before subscribing to patch stream.
* @param {number} [sinceSeq]
* @return {Promise<SnapshotResponseMessage>}
*/
requestSnapshot(sinceSeq = 0)
{
this._assertActive()
const requestId = crypto.randomUUID()
this._snapshotReady = false
return new Promise((resolve, reject) => {
const timer = setTimeout(() => {
this._pendingSnapshots.delete(requestId)
reject(new Error('BrazenEventBus: snapshot request timed out'))
}, this._snapshotTimeoutMs)
this._pendingSnapshots.set(requestId, {resolve, reject, timer})
this._postToChannel({
kind: 'snapshot-request',
tabId: this.tabId,
requestId,
sinceSeq,
})
})
}
/**
* Coordinator tab (or post-acquire): do not gate remote patches; flush any buffered stream.
* @return {void}
*/
markSnapshotReady()
{
this._assertActive()
if (this._snapshotReady) {
this._releaseBufferedPatches()
return
}
this._snapshotReady = true
this._releaseBufferedPatches()
}
/**
* Coordinator: respond to snapshot-request.
* @param {SnapshotRequestMessage} request
* @return {Promise<void>}
*/
async handleSnapshotRequest(request)
{
this._assertActive()
if (!this._snapshotResponder) {
throw new Error('BrazenEventBus: no snapshot responder registered')
}
const result = await this._snapshotResponder(request)
this.publish({
kind: 'snapshot-response',
tabId: this.tabId,
requestId: request.requestId,
snapshot: structuredClone(result.snapshot),
snapshotSeq: result.snapshotSeq,
catchUpPatches: result.catchUpPatches ? structuredClone(result.catchUpPatches) : undefined,
})
}
/**
* Tear down channel + listeners.
*/
dispose()
{
if (this._disposed) {
return
}
this._disposed = true
for (const pending of this._pendingSnapshots.values()) {
clearTimeout(pending.timer)
pending.reject(new Error('BrazenEventBus disposed'))
}
this._pendingSnapshots.clear()
this._localListeners.clear()
this._subscribers.clear()
this._deliveryQueue = []
this._patchBuffer = []
if (this._gapResyncTimer) {
clearTimeout(this._gapResyncTimer)
this._gapResyncTimer = null
}
if (this._channel) {
this._channel.onmessage = null
this._channel.close()
this._channel = null
}
}
// -------------------------------------------------------------------------
// Private class methods
// -------------------------------------------------------------------------
/**
* @private
*/
_assertActive()
{
if (this._disposed) {
throw new Error('BrazenEventBus disposed')
}
}
/**
* @private
*/
_openChannel()
{
this._channel = this._createBroadcastChannel(this.channelName)
this._channel.onmessage = (event) => {
const message = /** @type {BusMessage} */ (event.data)
if (!message || message.tabId === this.tabId) {
return
}
this._receiveRemote(message)
}
}
/**
* @param {BusMessage} message
* @private
*/
_stamp(message)
{
return structuredClone({
...message,
tabId: message.tabId ?? this.tabId,
ts: message.ts ?? Date.now(),
})
}
/**
* @param {BusMessage} message
* @private
*/
_emitLocal(message)
{
for (const listener of this._localListeners) {
listener(message)
}
this._enqueueForSubscribers(message, {bypassSnapshotGate: true})
}
/**
* @param {BusMessage} message
* @private
*/
_postToChannel(message)
{
if (this._channel) {
this._channel.postMessage(structuredClone(message))
}
}
/**
* @param {BusMessage} message
* @private
*/
_receiveRemote(message)
{
if (message.kind === 'snapshot-response') {
this._tryResolveSnapshot(/** @type {SnapshotResponseMessage} */ (message))
}
if (message.kind === 'snapshot-request' && this._snapshotResponder) {
void this.handleSnapshotRequest(/** @type {SnapshotRequestMessage} */ (message))
}
this._enqueueForSubscribers(message)
}
/**
* @param {BusMessage} message
* @param {{bypassSnapshotGate?: boolean}} [options]
* @private
*/
_enqueueForSubscribers(message, options = {})
{
if (!this._snapshotReady && message.kind === 'patch' && !options.bypassSnapshotGate) {
this._patchBuffer.push(/** @type {PatchMessage} */ (message))
return
}
if (this._deliveryQueue.length >= MAX_DELIVERY_QUEUE) {
this._deliveryQueue.shift()
}
this._deliveryQueue.push(message)
this._drainDeliveryQueue()
}
/**
* @private
*/
_drainDeliveryQueue()
{
if (this._delivering) {
return
}
this._delivering = true
try {
while (this._deliveryQueue.length > 0) {
const index = this._findNextDeliverableIndex()
if (index < 0) {
this._maybeScheduleGapResync()
break
}
const message = this._deliveryQueue.splice(index, 1)[0]
this._deliverToSubscribers(message)
}
} finally {
this._delivering = false
}
}
/**
* @return {number}
* @private
*/
_findNextDeliverableIndex()
{
for (let index = 0; index < this._deliveryQueue.length; index += 1) {
const message = this._deliveryQueue[index]
if (message.kind !== 'patch') {
return index
}
const patch = /** @type {PatchMessage} */ (message)
if (patch.seq <= this._nextExpectedPatchSeq) {
return index
}
if (patch.seq === this._nextExpectedPatchSeq + 1) {
return index
}
}
return -1
}
/**
* @private
*/
_maybeScheduleGapResync()
{
if (this._snapshotResponder || this._gapResyncInFlight || this._gapResyncTimer) {
return
}
let minGapSeq = Infinity
for (const message of this._deliveryQueue) {
if (message.kind === 'patch') {
let seq = /** @type {PatchMessage} */ (message).seq
if (seq > this._nextExpectedPatchSeq + 1) {
minGapSeq = Math.min(minGapSeq, seq)
}
}
}
if (minGapSeq === Infinity) {
return
}
this._gapResyncTimer = setTimeout(() => {
this._gapResyncTimer = null
this._gapResyncInFlight = true
void this.requestSnapshot(this._nextExpectedPatchSeq)
.then(() => {
this._deliveryQueue = []
this._patchBuffer = []
})
.catch(() => {})
.finally(() => {
this._gapResyncInFlight = false
})
}, GAP_RESYNC_WAIT_MS)
}
/**
* @param {BusMessage} message
* @private
*/
_deliverToSubscribers(message)
{
if (message.kind === 'patch') {
const patch = /** @type {PatchMessage} */ (message)
if (patch.seq <= this._nextExpectedPatchSeq) {
return
}
this._nextExpectedPatchSeq = patch.seq
}
for (const subscriber of this._subscribers) {
subscriber(message)
}
}
/**
* @param {SnapshotResponseMessage} message
* @private
*/
_tryResolveSnapshot(message)
{
const pending = this._pendingSnapshots.get(message.requestId)
if (!pending) {
return
}
clearTimeout(pending.timer)
this._pendingSnapshots.delete(message.requestId)
this._snapshotReady = true
this._nextExpectedPatchSeq = message.snapshotSeq
pending.resolve(message)
this._releaseBufferedPatches()
}
/**
* @private
*/
_releaseBufferedPatches()
{
if (this._patchBuffer.length === 0) {
return
}
this._patchBuffer.sort((left, right) => left.seq - right.seq)
const buffered = this._patchBuffer
this._patchBuffer = []
for (const patch of buffered) {
if (patch.seq > this._nextExpectedPatchSeq) {
this._deliveryQueue.push(patch)
}
}
this._drainDeliveryQueue()
}
}
globalThis.BrazenEventBus = BrazenEventBus
// -------------------------------------------------------------------------
// JobRegistry
// -------------------------------------------------------------------------
/** @typedef {import('./reactor-core.spec.md').JobDescriptor} JobDescriptor */
/** @type {string} */
const JOB_TYPE_RESOLVE = 'resolve'
/** @type {string} */
const JOB_TYPE_DOWNLOAD = 'download'
/** @type {string} Per-tab UI / chrome reactions — runnable on any tab. */
const JOB_SCOPE_LOCAL = 'local'
/** @type {string} Coordinator-only mutations / cross-tab pipeline work. */
const JOB_SCOPE_COORDINATOR = 'coordinator'
/**
* @param {import('./reactor-core.spec.md').JobDescriptor|undefined} descriptor
* @return {string}
*/
function jobDescriptorScope(descriptor)
{
return descriptor?.scope === JOB_SCOPE_LOCAL ? JOB_SCOPE_LOCAL : JOB_SCOPE_COORDINATOR
}
class BrazenJobRegistry
{
// -------------------------------------------------------------------------
// Protected class variables
// -------------------------------------------------------------------------
/** @type {Map<string, JobDescriptor>} */
_descriptors = new Map()
// -------------------------------------------------------------------------
// Public class methods
// -------------------------------------------------------------------------
/**
* Register a job type descriptor.
* @template TPayload
* @template TResult
* @param {string} type
* @param {JobDescriptor<TPayload, TResult>} descriptor
*/
register(type, descriptor)
{
if (!type || typeof type !== 'string') {
throw new Error('BrazenJobRegistry.register: type must be a non-empty string')
}
if (!descriptor || typeof descriptor.run !== 'function') {
throw new Error(`BrazenJobRegistry.register: "${type}" descriptor.run must be a function`)
}
this._descriptors.set(type, descriptor)
}
/**
* @param {string} type
* @return {JobDescriptor | undefined}
*/
get(type)
{
return this._descriptors.get(type)
}
/**
* @return {string[]}
*/
types()
{
return [...this._descriptors.keys()]
}
}
globalThis.BrazenJobRegistry = BrazenJobRegistry
globalThis.JOB_TYPE_RESOLVE = JOB_TYPE_RESOLVE
globalThis.JOB_TYPE_DOWNLOAD = JOB_TYPE_DOWNLOAD
globalThis.JOB_SCOPE_LOCAL = JOB_SCOPE_LOCAL
globalThis.JOB_SCOPE_COORDINATOR = JOB_SCOPE_COORDINATOR
// -------------------------------------------------------------------------
// Scheduler
// -------------------------------------------------------------------------
/**
* @typedef {object} SchedulerOptions
* @property {BrazenJobRegistry} registry
* @property {object} kernel
* @property {() => number} [kernel.nextSeq]
* @property {AbortSignal} [kernel.abortSignal]
* @property {() => unknown} [kernel.readState]
* @property {(path: string) => unknown} [kernel.read]
* @property {boolean} [linkQueues]
*/
/**
* @typedef {object} JobRecord
* @property {string} type
* @property {string} key
* @property {unknown} payload
* @property {number} seq
* @property {number} priority
* @property {AbortSignal} abortSignal
* @property {number|null} [originCauseId]
* @property {string|null} [originCauseLabel]
*/
/** @type {Set<string>} */
const TERMINAL_JOB_STATUSES = new Set(['done', 'failed', 'skipped', 'cancelled', 'duplicate'])
/** Maximum retained terminal rows before pruning unreferenced keys. */
const TERMINAL_MAP_CAP = 512
/**
* @typedef {'done' | 'failed' | 'cancelled' | 'skipped' | 'duplicate'} JobTerminalStatus
*/
/**
* @param {string} type
* @param {string} key
* @return {string}
*/
function pendingMapKey(type, key)
{
return `${type}\0${key}`
}
/**
* @param {string} type
* @param {string} key
* @return {string}
*/
function terminalMapKey(type, key)
{
return `${type}:${key}`
}
/**
* @param {AbortSignal} parent
* @return {AbortController}
*/
function createScopedAbortController(parent)
{
let controller = new AbortController()
if (parent.aborted) {
controller.abort(parent.reason)
return controller
}
let onAbort = () => {
controller.abort(parent.reason)
}
parent.addEventListener('abort', onAbort, {once: true})
controller.signal.addEventListener('abort', () => {
parent.removeEventListener('abort', onAbort)
}, {once: true})
return controller
}
class BrazenScheduler
{
/**
* @param {SchedulerOptions} options
*/
constructor(options)
{
if (!options?.registry) {
throw new Error('BrazenScheduler: registry is required')
}
if (!options?.kernel) {
throw new Error('BrazenScheduler: kernel is required')
}
this._registry = options.registry
this._kernel = options.kernel
this._linkQueues = options.linkQueues === true
this._scopeController = createScopedAbortController(
options.kernel.abortSignal ?? new AbortController().signal,
)
/** @type {Map<string, JobRecord>} */
this._pending = new Map()
/** @type {Map<string, Promise<void>>} */
this._running = new Map()
/** @type {Map<string, JobTerminalStatus>} */
this._terminal = new Map()
/** @type {Map<string, AbortController>} */
this._jobControllers = new Map()
/** @type {Map<string, number>} */
this._runningCountByType = new Map()
this._localSeq = 0
this._idleWaiters = []
}
// -------------------------------------------------------------------------
// Public class methods
// -------------------------------------------------------------------------
/**
* Enqueue or coalesce a job; assigns seq from kernel when available.
* @param {string} type
* @param {unknown} payload
* @param {{ key?: string, priority?: number }} [options]
*/
schedule(type, payload, options = {})
{
let descriptor = this._registry.get(type)
if (!descriptor) {
throw new Error(`BrazenScheduler.schedule: unregistered job type "${type}"`)
}
let key = options.key
if (key == null) {
if (typeof descriptor.coalesceKey === 'function') {
key = descriptor.coalesceKey(payload)
} else {
key = `${type}:${JSON.stringify(payload)}`
}
}
key = String(key)
let mapKey = pendingMapKey(type, key)
let priority = options.priority ?? descriptor.priority ?? 0
let seq = this._nextSeq()
let profiler = brazenReactorProfiler()
let originCauseId = profiler.activeCauseId()
let originCauseLabel = profiler.activeCauseLabel()
if (this._runningCountForType(type) > 0 || profiler.causeChainHasJobType(type, originCauseId)) {
profiler.noteScheduledReentrant(type, originCauseLabel, originCauseId)
}
if (this._pending.has(mapKey)) {
let pending = this._pending.get(mapKey)
let incoming = {
type,
key,
payload,
seq,
priority: Math.max(pending.priority, priority),
abortSignal: pending.abortSignal,
originCauseId,
originCauseLabel,
}
let merge = descriptor.merge ?? ((_, next) => next)
let merged = merge(pending, incoming)
if (!merged) {
profiler.mark('scheduler', 'schedule-merge-drop', {key: `${type}:${key}`})
return
}
merged.priority = Math.max(pending.priority, priority)
merged.seq = Math.min(pending.seq, seq)
merged.abortSignal = pending.abortSignal
merged.originCauseId = pending.originCauseId ?? originCauseId
merged.originCauseLabel = pending.originCauseLabel ?? originCauseLabel
this._pending.set(mapKey, merged)
profiler.mark('scheduler', 'schedule-coalesce', {
key: `${type}:${key}`,
data: {cause: merged.originCauseLabel},
})
return
}
let controller = createScopedAbortController(this._scopeController.signal)
this._jobControllers.set(mapKey, controller)
this._pending.set(mapKey, {
type,
key,
payload,
seq,
priority,
abortSignal: controller.signal,
originCauseId,
originCauseLabel,
})
profiler.mark('scheduler', 'schedule-enqueue', {
key: `${type}:${key}`,
data: {priority: priority, cause: originCauseLabel},
})
}
/** Spec alias for schedule. */
enqueue(type, payload, options = {})
{
this.schedule(type, payload, options)
}
/**
* Cancel pending work and abort in-flight execution for (type, key).
* @param {string} type
* @param {string} key
* @param {string} [reason]
*/
cancelPending(type, key, reason)
{
let mapKey = pendingMapKey(type, String(key))
let hadPending = this._pending.has(mapKey)
this._pending.delete(mapKey)
let controller = this._jobControllers.get(mapKey)
this._jobControllers.delete(mapKey)
if (controller && !controller.signal.aborted) {
controller.abort(reason ?? 'cancelled')
}
if (hadPending || this._running.has(mapKey)) {
this._markTerminal(type, String(key), 'cancelled')
}
}
/** Spec alias for cancelPending. */
cancel(type, key, reason)
{
this.cancelPending(type, key, reason)
}
/**
* Enqueue a local-scope job (per-tab UI lane). Validates descriptor.scope === 'local'.
* @param {string} type
* @param {unknown} payload
* @param {{ key?: string, priority?: number }} [options]
*/
scheduleLocal(type, payload, options = {})
{
let descriptor = this._registry.get(type)
if (!descriptor) {
throw new Error(`BrazenScheduler.scheduleLocal: unregistered job type "${type}"`)
}
if (jobDescriptorScope(descriptor) !== JOB_SCOPE_LOCAL) {
throw new Error(`BrazenScheduler.scheduleLocal: "${type}" is not a local-scope job`)
}
this.schedule(type, payload, options)
}
/**
* Abort all pending and in-flight jobs (coordinator hand-off / tab teardown).
* @param {string} [reason]
*/
cancelAll(reason)
{
for (let mapKey of [...this._pending.keys()]) {
let job = this._pending.get(mapKey)
if (job) {
this.cancelPending(job.type, job.key, reason)
}
}
for (let controller of this._jobControllers.values()) {
if (!controller.signal.aborted) {
controller.abort(reason ?? 'cancelled')
}
}
}
/**
* Pump the ready set until idle or the scheduler scope is aborted.
* @return {Promise<void>}
*/
async runReady()
{
let profiler = brazenReactorProfiler()
let pumpDepth = profiler.enter('scheduler', 'runReady')
let stopPump = profiler.time('scheduler', 'runReady')
let syncSpins = 0
let yields = 0
while (!this._isIdle() && !this._scopeController.signal.aborted) {
let started = this._startReadyJobs()
if (started === 0) {
if (this._running.size > 0) {
await this._waitForAnyRunning()
} else {
break
}
} else {
syncSpins += 1
if (syncSpins >= 8) {
syncSpins = 0
yields += 1
await new Promise((resolve) => setTimeout(resolve, 0))
}
}
}
await this._waitForAllRunning()
stopPump()
profiler.mark('scheduler', 'runReady-end', {
data: {syncSpins, yields, pending: this._pending.size, running: this._running.size},
})
profiler.exit('scheduler', 'runReady')
}
/** Spec alias for runReady. */
async runUntilIdle()
{
return this.runReady()
}
/**
* Last terminal outcome for (type, key) in this scheduler session (for stall detection).
* @param {string} type
* @param {string} key
* @return {JobTerminalStatus|null}
*/
getTerminalStatus(type, key)
{
return this._terminal.get(terminalMapKey(type, String(key))) ?? null
}
/**
* @param {string} [type]
* @return {number}
*/
pendingCount(type)
{
if (!type) {
return this._pending.size
}
let count = 0
for (let job of this._pending.values()) {
if (job.type === type) {
count += 1
}
}
return count
}
// -------------------------------------------------------------------------
// Private class methods
// -------------------------------------------------------------------------
/**
* @return {number}
* @private
*/
_nextSeq()
{
if (typeof this._kernel.nextSeq === 'function') {
return this._kernel.nextSeq()
}
this._localSeq += 1
return this._localSeq
}
/**
* @return {boolean}
* @private
*/
_isIdle()
{
return this._pending.size === 0 && this._running.size === 0
}
/**
* @return {Promise<void>}
* @private
*/
_waitForAnyRunning()
{
if (this._running.size === 0) {
return Promise.resolve()
}
return Promise.race([...this._running.values()])
}
/**
* @return {Promise<void>}
* @private
*/
_waitForAllRunning()
{
if (this._running.size === 0) {
return Promise.resolve()
}
return Promise.all([...this._running.values()])
}
/**
* @private
*/
_notifyIdleWaiter()
{
if (this._isIdle() && this._idleWaiters.length) {
let waiters = this._idleWaiters.splice(0)
for (let resolve of waiters) {
resolve()
}
}
}
/**
* @param {string} type
* @param {string} key
* @param {JobTerminalStatus} status
* @private
*/
_markTerminal(type, key, status)
{
this._terminal.set(terminalMapKey(type, key), status)
this._pruneTerminalMap()
}
/**
* @private
*/
_pruneTerminalMap()
{
if (this._terminal.size <= TERMINAL_MAP_CAP) {
return
}
/** @type {Set<string>} */
let referenced = new Set()
for (let job of this._pending.values()) {
let descriptor = this._registry.get(job.type)
for (let depType of descriptor?.deps ?? []) {
referenced.add(terminalMapKey(depType, job.key))
}
}
for (let key of this._terminal.keys()) {
if (!referenced.has(key)) {
this._terminal.delete(key)
}
if (this._terminal.size <= TERMINAL_MAP_CAP) {
break
}
}
}
/**
* @param {JobRecord} job
* @return {boolean}
* @private
*/
_depsSatisfied(job)
{
let descriptor = this._registry.get(job.type)
let deps = descriptor?.deps ?? []
for (let depType of deps) {
let depTerminal = this._terminal.get(terminalMapKey(depType, job.key))
if (depTerminal && TERMINAL_JOB_STATUSES.has(depTerminal)) {
continue
}
if (this._pending.has(pendingMapKey(depType, job.key)) ||
this._running.has(pendingMapKey(depType, job.key))) {
return false
}
return false
}
return true
}
/**
* @param {string} type
* @return {number}
* @private
*/
_concurrencyLimit(type)
{
let descriptor = this._registry.get(type)
return descriptor?.concurrency ?? 1
}
/**
* @param {string} type
* @return {number}
* @private
*/
_runningCountForType(type)
{
return this._runningCountByType.get(type) ?? 0
}
/**
* @return {JobRecord[]}
* @private
*/
_collectReadyJobs()
{
/** @type {JobRecord[]} */
let ready = []
let isCoordinator = typeof this._kernel.isCoordinator === 'function' ?
this._kernel.isCoordinator() :
true
for (let job of this._pending.values()) {
if (!this._depsSatisfied(job)) {
continue
}
let descriptor = this._registry.get(job.type)
if (jobDescriptorScope(descriptor) === JOB_SCOPE_COORDINATOR && !isCoordinator) {
continue
}
if (this._runningCountForType(job.type) >= this._concurrencyLimit(job.type)) {
continue
}
ready.push(job)
}
ready.sort((a, b) => {
if (this._linkQueues) {
let rank = (type) => {
if (type === 'resolve') {
return 0
}
if (type === 'download') {
return 1
}
return 2
}
let rankDiff = rank(a.type) - rank(b.type)
if (rankDiff !== 0) {
return rankDiff
}
}
if (b.priority !== a.priority) {
return b.priority - a.priority
}
return a.seq - b.seq
})
/** @type {JobRecord[]} */
let selected = []
/** @type {Map<string, number>} */
let selectedCounts = new Map()
for (let job of ready) {
let count = selectedCounts.get(job.type) ?? 0
if (count >= this._concurrencyLimit(job.type)) {
continue
}
selected.push(job)
selectedCounts.set(job.type, count + 1)
}
return selected
}
/**
* @return {number}
* @private
*/
_startReadyJobs()
{
let ready = this._collectReadyJobs()
let profiler = brazenReactorProfiler()
for (let job of ready) {
let mapKey = pendingMapKey(job.type, job.key)
this._pending.delete(mapKey)
profiler.mark('scheduler', 'job-start', {key: `${job.type}:${job.key}`})
let runPromise = this._executeJob(job)
this._running.set(mapKey, runPromise)
this._runningCountByType.set(job.type, this._runningCountForType(job.type) + 1)
runPromise.finally(() => {
this._running.delete(mapKey)
this._runningCountByType.set(
job.type,
Math.max(0, this._runningCountForType(job.type) - 1),
)
this._jobControllers.delete(mapKey)
this._notifyIdleWaiter()
})
}
if (ready.length) {
profiler.mark('scheduler', 'jobs-started', {data: {count: ready.length}})
}
return ready.length
}
/**
* @param {JobRecord} job
* @return {Promise<void>}
* @private
*/
async _executeJob(job)
{
let descriptor = this._registry.get(job.type)
let profiler = brazenReactorProfiler()
let jobKey = `${job.type}:${job.key}`
return profiler.withCause({
kind: 'job',
label: jobKey,
parentId: job.originCauseId ?? undefined,
}, async () => {
let stop = profiler.time('scheduler', 'job-run', {key: jobKey})
let terminal = 'failed'
if (!descriptor) {
this._markTerminal(job.type, job.key, 'failed')
stop()
profiler.mark('scheduler', 'job-end', {key: jobKey, data: {terminal: 'failed'}})
return
}
if (job.abortSignal.aborted) {
this._markTerminal(job.type, job.key, 'cancelled')
stop()
profiler.mark('scheduler', 'job-end', {key: jobKey, data: {terminal: 'cancelled'}})
return
}
let readState = () => {
if (typeof this._kernel.readState === 'function') {
return this._kernel.readState()
}
return {}
}
if (typeof descriptor.reassess === 'function') {
let reassessed = descriptor.reassess(job, readState)
if (reassessed == null) {
this._markTerminal(job.type, job.key, 'skipped')
stop()
profiler.mark('scheduler', 'job-end', {key: jobKey, data: {terminal: 'skipped'}})
return
}
job = {...job, ...reassessed, abortSignal: job.abortSignal}
}
/** @type {import('./reactor-core.spec.md').JobContext} */
let ctx = {
kernel: this._kernel,
repos: this._kernel.repos ?? {},
abortSignal: job.abortSignal,
spawn: (type, spawnPayload) => {
if (!profiler.noteSpawn(job.key, type)) {
return
}
this.schedule(type, spawnPayload, {key: job.key})
},
read: (path) => {
if (typeof this._kernel.read === 'function') {
return this._kernel.read(path)
}
let state = readState()
if (state && typeof state === 'object' && path in state) {
return state[path]
}
return undefined
},
yield: () => new Promise((resolve) => setTimeout(resolve, 0)),
}
try {
await descriptor.run(ctx, job.payload)
if (job.abortSignal.aborted) {
terminal = 'cancelled'
this._markTerminal(job.type, job.key, 'cancelled')
} else {
terminal = 'done'
this._markTerminal(job.type, job.key, 'done')
}
} catch (error) {
if (job.abortSignal.aborted) {
terminal = 'cancelled'
this._markTerminal(job.type, job.key, 'cancelled')
} else if (error && typeof error === 'object' && error.brazenSchedulerSkip) {
terminal = 'skipped'
this._markTerminal(job.type, job.key, 'skipped')
} else {
terminal = 'failed'
this._markTerminal(job.type, job.key, 'failed')
}
} finally {
stop()
profiler.mark('scheduler', 'job-end', {key: jobKey, data: {terminal}})
}
})
}
}
globalThis.BrazenScheduler = BrazenScheduler
// -------------------------------------------------------------------------
// Kernel
// -------------------------------------------------------------------------
const KERNEL_SEQ_MAX_SAFE = Number.MAX_SAFE_INTEGER - 1
/**
* @param {string} scriptPrefix
* @return {string}
*/
function coordinatorLockName(scriptPrefix)
{
return `brazen-${String(scriptPrefix ?? '').trim()}-coordinator`
}
class BrazenKernel
{
// -------------------------------------------------------------------------
// Static public methods
// -------------------------------------------------------------------------
/**
* @param {string} scriptPrefix
* @return {string}
*/
static coordinatorLockName(scriptPrefix)
{
return coordinatorLockName(scriptPrefix)
}
// -------------------------------------------------------------------------
// Protected class variables
// -------------------------------------------------------------------------
/** @type {string} */
_scriptPrefix = ''
/** @type {string} */
_lockName = ''
/** @type {object|null} */
_signals = null
/** @type {object|null} */
_bus = null
/** @type {object|null} */
_repos = null
/** @type {object|null} */
_scheduler = null
/** @type {(() => void)|null} */
_onCoordinatorAcquired = null
/** @type {((reason: 'steal'|'release'|'abort') => void)|null} */
_onCoordinatorLost = null
/** @type {((command: object, seq: number, patches: object[]) => void)|null} */
_onCommandApplied = null
/** @type {AbortController} */
_abortController = new AbortController()
/** @type {number} */
_seq = 0
/** @type {boolean} */
_isCoordinator = false
/** @type {boolean} */
_started = false
/** @type {boolean} */
_hydrated = false
/** @type {boolean} */
_documentSuspended = false
/** @type {(() => void)|null} */
_releaseLock = null
/** @type {'steal'|'release'|'abort'|null} */
_lossReason = null
/** @type {Promise<void>|null} */
_lockLoopPromise = null
/** @type {(() => void)|null} */
_busUnsubscribe = null
/** @type {boolean} */
_lifecycleBound = false
/** @type {Map<string, unknown>} */
_authoritativeValues = new Map()
/** @type {{seq: number, patches: object[]}[]} */
_committedPatches = []
/** @type {Promise<void>|null} */
_hydratePromise = null
/** @type {boolean} */
_schedulerPumpScheduled = false
/** @type {boolean} */
_schedulerPumpRunning = false
/** @type {ReturnType<typeof setTimeout>|null} */
_schedulerPumpWatchdog = null
/** Maximum ms before a stuck runReady() releases the pump flag. */
static SCHEDULER_PUMP_WATCHDOG_MS = 30000
/**
* @param {object} options
* @param {string} options.scriptPrefix
* @param {object} options.signals
* @param {object} options.bus
* @param {object} [options.repos]
* @param {object} options.scheduler
* @param {() => void} [options.onCoordinatorAcquired]
* @param {(reason: 'steal'|'release'|'abort') => void} [options.onCoordinatorLost]
* @param {(command: object, seq: number, patches: object[]) => void} [options.onCommandApplied]
*/
constructor(options = {})
{
this._scriptPrefix = String(options.scriptPrefix ?? '').trim()
this._lockName = coordinatorLockName(this._scriptPrefix)
this._signals = options.signals ?? null
this._bus = options.bus ?? null
this._repos = options.repos ?? null
this._scheduler = options.scheduler ?? null
this._onCoordinatorAcquired = typeof options.onCoordinatorAcquired === 'function'
? options.onCoordinatorAcquired
: null
this._onCoordinatorLost = typeof options.onCoordinatorLost === 'function'
? options.onCoordinatorLost
: null
this._onCommandApplied = typeof options.onCommandApplied === 'function'
? options.onCommandApplied
: null
}
// -------------------------------------------------------------------------
// Public getters
// -------------------------------------------------------------------------
/** @return {AbortSignal} */
get abortSignal()
{
return this._abortController.signal
}
// -------------------------------------------------------------------------
// Public class methods
// -------------------------------------------------------------------------
/**
* Start lock acquisition loop + bus command handler. Idempotent.
* @return {Promise<void>}
*/
async start()
{
if (this._started) {
return
}
this._started = true
this._documentSuspended = false
this._bindLifecycle()
this._ensureBusSubscription()
this._lockLoopPromise = this._runCoordinatorLockLoop()
}
/**
* Release lock gracefully (pagehide). Aborts in-flight jobs.
* @param {'pagehide'|'manual'} [reason]
* @return {Promise<void>}
*/
async stop(reason = 'manual')
{
if (reason === 'pagehide') {
this._documentSuspended = true
this._lossReason = 'abort'
} else {
this._lossReason = 'release'
}
this._started = false
if (this._releaseLock) {
let release = this._releaseLock
this._releaseLock = null
release()
}
if (this._lockLoopPromise) {
try {
await this._lockLoopPromise
} catch (e) {
// lock loop may reject after steal
}
this._lockLoopPromise = null
}
}
/**
* @return {boolean}
*/
isCoordinator()
{
return this._isCoordinator
}
/**
* @param {{steal?: boolean, ifAvailable?: boolean}} [options]
* @return {Promise<boolean>}
*/
async requestCoordinatorRole(options = {})
{
if (this._documentSuspended || !this._locksAvailable()) {
return false
}
if (this.isCoordinator()) {
return true
}
let steal = options.steal === true
let ifAvailable = options.ifAvailable === true
if (steal) {
return this._requestLock({steal: true})
}
if (ifAvailable) {
return this._requestLock({ifAvailable: true})
}
return this._requestLock({ifAvailable: true})
}
/**
* @return {number}
*/
getSeq()
{
return this._seq
}
/**
* @param {object} command
* @return {Promise<void>}
*/
async dispatch(command)
{
if (!this.isCoordinator()) {
throw new Error('BrazenKernel.dispatch requires coordinator role')
}
await this.handleCommand({
kind: 'command',
tabId: this._bus?.tabId ?? 'local',
command,
})
}
/**
* Coordinator: broadcast patches after IDB already committed (patch-only — no write-through).
* @param {object} _command audit stub (dm-state-mutation / dm-state-progress)
* @param {object[]} patches `{path, value}` entries; seq/version assigned here
* @return {Promise<void>}
*/
async commitStatePatches(_command, patches)
{
if (!this.isCoordinator() || !patches?.length) {
return
}
let profiler = brazenReactorProfiler()
return profiler.withCause({
kind: 'state-progress',
label: _command?.type ?? 'state-progress',
detail: {patchCount: patches.length},
}, async () => {
if (!this._hydrated) {
await this.hydrate()
}
let seq = this._assignSeq()
let stamped = this._stampPatchesForCommit(patches, seq)
this._recordCommittedPatch(seq, stamped)
this._publishPatch(seq, stamped)
})
}
/**
* Enqueue a local-scope scheduler job on this tab and pump the ready set.
* @param {string} type
* @param {unknown} payload
* @param {{ key?: string, priority?: number }} [options]
*/
scheduleLocal(type, payload, options = {})
{
if (!this._scheduler || typeof this._scheduler.scheduleLocal !== 'function') {
return
}
this._scheduler.scheduleLocal(type, payload, options)
this._ensureSchedulerPump()
}
/**
* Hydrate atoms from IDB + serve snapshot responses.
* @return {Promise<void>}
*/
async hydrate()
{
if (this._hydratePromise) {
return this._hydratePromise
}
this._hydratePromise = this._doHydrate()
try {
await this._hydratePromise
} finally {
this._hydratePromise = null
}
}
/**
* Apply command: validate → mutate atoms → schedule jobs → IDB write-through → broadcast patch.
* @param {object} message
* @return {Promise<void>}
*/
async handleCommand(message)
{
if (!this.isCoordinator() || !message || message.kind !== 'command') {
return
}
let command = message.command
if (!command || typeof command.type !== 'string') {
return
}
let profiler = brazenReactorProfiler()
return profiler.withCause({kind: 'command', label: command.type}, async () => {
let stop = profiler.time('kernel', 'handleCommand', {key: command.type})
if (command.type === 'request-coordinator-steal') {
await this.requestCoordinatorRole({steal: true})
stop()
return
}
if (!this._hydrated) {
await this.hydrate()
}
let seq = this._assignSeq()
let patches = this._mutateAtomsForCommand(command, seq)
await this._invokeReposWriteThrough(command, patches)
patches = await this._mergePostWritePatches(command, seq, patches)
this._scheduleJobsForCommand(command, seq)
this._ensureSchedulerPump()
this._recordCommittedPatch(seq, patches)
this._publishPatch(seq, patches)
this._onCommandApplied?.(command, seq, patches)
stop()
profiler.mark('kernel', 'handleCommand-end', {
key: command.type,
data: {seq, patchCount: patches.length},
})
})
}
/**
* Drain scheduler jobs (local on any tab; coordinator-scope on coordinator only).
* @private
*/
_ensureSchedulerPump()
{
if (!this._scheduler || typeof this._scheduler.runReady !== 'function') {
return
}
if (this._schedulerPumpScheduled) {
return
}
this._schedulerPumpScheduled = true
brazenReactorProfiler().mark('kernel', 'pump-scheduled')
queueMicrotask(() => {
this._schedulerPumpScheduled = false
if (this._schedulerPumpRunning) {
brazenReactorProfiler().mark('kernel', 'pump-skipped-running')
return
}
this._startSchedulerPump()
})
}
/**
* @private
*/
_startSchedulerPump()
{
let profiler = brazenReactorProfiler()
this._schedulerPumpRunning = true
profiler.mark('kernel', 'pump-start', {data: {pending: this._scheduler.pendingCount?.() ?? 0}})
if (this._schedulerPumpWatchdog) {
clearTimeout(this._schedulerPumpWatchdog)
}
let settled = false
let finishPump = (reason = 'done') => {
if (settled) {
return
}
settled = true
if (this._schedulerPumpWatchdog) {
clearTimeout(this._schedulerPumpWatchdog)
this._schedulerPumpWatchdog = null
}
this._schedulerPumpRunning = false
profiler.mark('kernel', 'pump-finish', {
data: {
reason,
pending: this._scheduler.pendingCount?.() ?? 0,
},
})
if (this._scheduler.pendingCount() > 0) {
this._ensureSchedulerPump()
}
}
this._schedulerPumpWatchdog = setTimeout(() => {
profiler._tripBreaker('kernel:pump-watchdog', '[BrazenReactor] scheduler pump watchdog fired', {
pending: this._scheduler.pendingCount?.() ?? 0,
})
finishPump('watchdog')
}, BrazenKernel.SCHEDULER_PUMP_WATCHDOG_MS)
void this._scheduler.runReady().
then(() => finishPump('done')).
catch(() => finishPump('error'))
}
/**
* After IDB write-through, stamp normative dm.state.* and pending count patches (v2 §5).
* @param {object} command
* @param {number} seq
* @param {object[]} patches
* @return {Promise<object[]>}
* @private
*/
async _mergePostWritePatches(command, seq, patches)
{
if (typeof this._repos?.kernelPostWritePatches !== 'function') {
return patches
}
let postRaw = await this._repos.kernelPostWritePatches(command)
if (!postRaw?.length) {
return patches
}
let stampedPost = this._stampPatchesForCommit(postRaw, seq)
let byPath = new Map(patches.map((patch) => [patch.path, patch]))
for (let patch of stampedPost) {
byPath.set(patch.path, patch)
}
return [...byPath.values()]
}
/**
* Coordinator: respond to snapshot-request.
* @param {object} request
* @return {Promise<void>}
*/
async handleSnapshotRequest(request)
{
if (!this.isCoordinator() || !request || request.kind !== 'snapshot-request') {
return
}
if (!this._hydrated) {
await this.hydrate()
}
let sinceSeq = Number(request.sinceSeq) || 0
let snapshot = this._buildSnapshotRecord()
let catchUpPatches = sinceSeq > 0
? this._flattenCatchUpPatches(sinceSeq, this._seq)
: undefined
if (typeof this._bus?.publish === 'function') {
this._bus.publish({
kind: 'snapshot-response',
tabId: this._bus.tabId,
requestId: request.requestId,
snapshot,
snapshotSeq: this._seq,
catchUpPatches,
})
}
}
// -------------------------------------------------------------------------
// Private class methods
// -------------------------------------------------------------------------
/**
* @return {Promise<void>}
* @private
*/
async _runCoordinatorLockLoop()
{
while (this._started && !this._documentSuspended) {
try {
await this._holdCoordinatorLock({})
} catch (e) {
if (this._isCoordinator) {
this._loseCoordinator('steal')
}
}
if (!this._started) {
break
}
}
}
/**
* @param {{steal?: boolean, ifAvailable?: boolean}} options
* @return {Promise<boolean>}
* @private
*/
async _requestLock(options = {})
{
if (!this._locksAvailable()) {
return false
}
return new Promise((resolveClaim) => {
void this._holdCoordinatorLock(options, resolveClaim)
})
}
/**
* @param {{steal?: boolean, ifAvailable?: boolean}} options
* @param {(claimed: boolean) => void} [onClaimed]
* @return {Promise<void>}
* @private
*/
async _holdCoordinatorLock(options = {}, onClaimed = null)
{
if (!this._locksAvailable()) {
onClaimed?.(false)
return
}
let lockOptions = {mode: 'exclusive'}
if (options.steal) {
lockOptions.steal = true
}
if (options.ifAvailable) {
lockOptions.ifAvailable = true
}
let claimedCalled = false
/** @param {boolean} claimed */
let claimOnce = (claimed) => {
if (claimedCalled) {
return
}
claimedCalled = true
onClaimed?.(claimed)
}
try {
await navigator.locks.request(this._lockName, lockOptions, async (lock) => {
if (!lock) {
claimOnce(false)
return
}
claimOnce(true)
this._becomeCoordinator()
try {
await new Promise((resolve) => {
this._releaseLock = resolve
})
} finally {
this._loseCoordinator(this._lossReason ?? 'release')
this._lossReason = null
this._releaseLock = null
}
})
} catch (error) {
if (this._isCoordinator) {
this._loseCoordinator('steal')
} else {
claimOnce(false)
}
throw error
}
}
/**
* @private
*/
_becomeCoordinator()
{
if (this._isCoordinator) {
return
}
this._abortController = new AbortController()
this._isCoordinator = true
if (!this._hydrated) {
void this.hydrate()
}
this._onCoordinatorAcquired?.()
}
/**
* @param {'steal'|'release'|'abort'} reason
* @private
*/
_loseCoordinator(reason)
{
if (!this._isCoordinator) {
return
}
this._isCoordinator = false
this._abortController.abort()
if (typeof this._scheduler?.cancelAll === 'function') {
this._scheduler.cancelAll(this.abortSignal.reason ?? reason)
}
this._onCoordinatorLost?.(reason)
}
/**
* @return {boolean}
* @private
*/
_locksAvailable()
{
return typeof navigator !== 'undefined' &&
navigator.locks != null &&
typeof navigator.locks.request === 'function'
}
/**
* @private
*/
_bindLifecycle()
{
if (this._lifecycleBound || typeof document === 'undefined') {
return
}
this._lifecycleBound = true
document.addEventListener('pagehide', (event) => {
void this.stop('pagehide')
})
document.addEventListener('pageshow', (event) => {
if (event.persisted) {
this._documentSuspended = false
void this.start()
}
})
}
/**
* @private
*/
_ensureBusSubscription()
{
if (this._busUnsubscribe || typeof this._bus?.subscribe !== 'function') {
return
}
this._busUnsubscribe = this._bus.subscribe((message) => {
if (message?.kind === 'command') {
void this.handleCommand(message)
} else if (message?.kind === 'snapshot-request') {
void this.handleSnapshotRequest(message)
}
})
}
/**
* @return {Promise<void>}
* @private
*/
async _doHydrate()
{
let stop = brazenReactorProfiler().time('kernel', 'hydrate')
if (typeof this._signals?.batch === 'function') {
await this._signals.batch(async () => {
await this._hydrateAuthoritativePathsFromRepos()
})
} else {
await this._hydrateAuthoritativePathsFromRepos()
}
this._hydrated = true
stop()
}
/**
* @private
*/
async _hydrateAuthoritativePathsFromRepos()
{
if (!this._repos) {
throw new Error('BrazenKernel.hydrate requires repos')
}
if (typeof this._repos.kernelHydrate !== 'function') {
throw new Error('BrazenKernel.hydrate requires repos.kernelHydrate')
}
let snapshot = await this._repos.kernelHydrate()
if (snapshot === undefined) {
throw new Error('BrazenKernel.hydrate: repos.kernelHydrate returned undefined')
}
if (snapshot && typeof snapshot === 'object') {
for (let [path, value] of Object.entries(snapshot)) {
this._authoritativeValues.set(path, structuredClone(value))
this._writeSignalPath(path, value, 0)
}
}
}
/**
* @param {object} command
* @param {number} seq
* @return {object[]}
* @private
*/
_stampPatchesForCommit(patches, seq)
{
let stamped = []
let writeOne = (path, value) => {
let cloned = structuredClone(value)
this._authoritativeValues.set(path, cloned)
let version = seq
if (typeof this._signals?.batch === 'function') {
this._signals.batch(() => {
version = this._writeSignalPath(path, cloned, seq) ?? seq
})
} else {
version = this._writeSignalPath(path, cloned, seq) ?? seq
}
stamped.push({path, version, value: cloned})
}
for (let patch of patches) {
if (!patch?.path) {
continue
}
writeOne(patch.path, patch.value)
}
return stamped
}
/**
* @param {object} command
* @param {number} seq
* @return {object[]}
* @private
*/
_mutateAtomsForCommand(command, seq)
{
let patches = []
let apply = (path, value) => {
let version = seq
this._authoritativeValues.set(path, structuredClone(value))
if (typeof this._signals?.batch === 'function') {
this._signals.batch(() => {
version = this._writeSignalPath(path, value, seq) ?? seq
})
} else {
version = this._writeSignalPath(path, value, seq) ?? seq
}
patches.push({path, version, value: structuredClone(value)})
}
switch (command.type) {
case 'toggle-paused':
apply('dm.state.paused', Boolean(command.payload?.paused))
break
case 'write-setting':
apply(`config.settings.${command.payload?.fieldKey}`, command.payload?.value)
break
case 'enqueue-download':
case 'dequeue-download':
case 'confirm-tag-discovery':
case 'skip-tag-discovery':
case 'clear-download-queue':
apply('kernel.lastCommand', {type: command.type, payload: command.payload ?? {}})
break
case 'config-save':
case 'config-sync':
apply('kernel.lastCommand', {type: command.type, payload: command.payload ?? {}})
break
case 'custom':
if (command.payload?.name !== 'pipeline-pump') {
apply(`kernel.custom.${command.payload?.name ?? 'unknown'}`, command.payload?.data)
}
break
default:
apply(`kernel.lastCommand`, {type: command.type, payload: command.payload ?? {}})
break
}
return patches
}
/**
* @param {string} path
* @param {unknown} value
* @param {number} fallbackVersion
* @return {number|undefined}
* @private
*/
_writeSignalPath(path, value, fallbackVersion)
{
if (typeof this._signals?.atom !== 'function') {
return fallbackVersion
}
let atomRef = this._signals.atom(path, value)
if (atomRef && typeof atomRef.write === 'function') {
return atomRef.write(value)
}
return fallbackVersion
}
/**
* @param {object} command
* @param {object[]} patches
* @return {Promise<void>}
* @private
*/
async _invokeReposWriteThrough(command, patches)
{
if (typeof this._repos?.kernelWriteThrough !== 'function') {
throw new Error('BrazenKernel: coordinator dispatch requires repos.kernelWriteThrough')
}
await this._repos.kernelWriteThrough(command, patches, this.getSeq())
}
/**
* @param {object} command
* @param {number} seq
* @private
*/
_scheduleJobsForCommand(command, seq)
{
if (typeof this._scheduler?.enqueue !== 'function') {
return
}
if (command.type === 'spawn-job') {
this._scheduler.enqueue(
command.payload?.jobType ?? 'unknown',
command.payload?.payload,
{key: command.payload?.key, priority: seq},
)
return
}
if (command.type === 'confirm-tag-discovery') {
this._scheduler.enqueue('confirm-tag-discovery', command.payload ?? {}, {
key: 'tag-discovery',
priority: seq,
})
return
}
if (command.type === 'skip-tag-discovery') {
this._scheduler.enqueue('skip-tag-discovery', command.payload ?? {}, {
key: 'tag-discovery',
priority: seq,
})
}
}
/**
* @param {number} seq
* @param {object[]} patches
* @private
*/
_recordCommittedPatch(seq, patches)
{
this._committedPatches.push({seq, patches: structuredClone(patches)})
if (this._committedPatches.length > 512) {
this._committedPatches.splice(0, this._committedPatches.length - 512)
}
}
/**
* @param {number} seq
* @param {object[]} patches
* @private
*/
_publishPatch(seq, patches)
{
brazenReactorProfiler().mark('kernel', 'publish-patch', {
key: String(seq),
data: {patchCount: patches?.length ?? 0},
})
if (typeof this._bus?.publish !== 'function') {
return
}
this._bus.publish({
kind: 'patch',
tabId: this._bus.tabId,
seq,
ts: Date.now(),
patches: structuredClone(patches),
})
}
/**
* @return {number}
* @private
*/
_assignSeq()
{
let next
if (typeof this._bus?.nextSeq === 'function') {
next = this._bus.nextSeq()
} else {
next = this._seq + 1
}
if (!Number.isFinite(next) || next <= this._seq) {
next = this._seq + 1
}
if (next > KERNEL_SEQ_MAX_SAFE) {
throw new Error('BrazenKernel seq overflow')
}
this._seq = next
return next
}
/**
* @return {Record<string, unknown>}
* @private
*/
_buildSnapshotRecord()
{
let snapshot = {}
for (let [path, value] of this._authoritativeValues.entries()) {
snapshot[path] = structuredClone(value)
}
return snapshot
}
/**
* @param {number} sinceSeq
* @param {number} snapshotSeq
* @return {object[]|undefined}
* @private
*/
_flattenCatchUpPatches(sinceSeq, snapshotSeq)
{
let out = []
for (let entry of this._committedPatches) {
if (entry.seq > sinceSeq && entry.seq <= snapshotSeq) {
out.push(...entry.patches.map((patch) => structuredClone(patch)))
}
}
return out.length ? out : undefined
}
}
globalThis.BrazenKernel = BrazenKernel
globalThis.coordinatorLockName = coordinatorLockName