Reactor Core: signals, event bus, job scheduler, and coordinator kernel
Script này sẽ không được không được cài đặt trực tiếp. Nó là một thư viện cho các script khác để bao gồm các chỉ thị meta
// @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