snxraven/bare-operating-system
1
1
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

normalize initd state modeling and stabilize synthetic initd PID mapping

Browse Source 
add IPC backpressure/drop telemetry and jittered swarm retry tiers
harden socket bridge timeout/lifecycle handling and replace recv polling with readiness signaling
improve shell background child isolation and worker offload fallback/output diagnostics
add var-log fast append path to reduce read-concat-write amplification
This commit is contained in:
Raven Scott
2026-04-26 06:57:07 -04:00
parent dfcd36dc58
commit 47da0b8531
9 changed files with 220 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/bare-os-booter/index.js
+50 -6
View File
@@ -6544,6 +6544,38 @@ async function executeKernel(disk, store, swarm, initSource) {
) {
if (wantPosixSocketFdBridge(shellEnv)) {
const st = this.bareOsPosixFdSimState
const bridgeNotify = (slot, kind) => {
const key = kind === 'accept' ? 'acceptWaiters' : 'recvWaiters'
const waiters = slot && Array.isArray(slot[key]) ? slot[key] : null
if (!waiters || waiters.length === 0) return
const pending = waiters.splice(0, waiters.length)
for (const resolve of pending) {
try {
resolve()
} catch {
/* ignore */
}
}
}
const bridgeWait = (slot, timeoutMs, kind) =>
new Promise((resolve) => {
const key = kind === 'accept' ? 'acceptWaiters' : 'recvWaiters'
if (!slot[key] || !Array.isArray(slot[key])) slot[key] = []
let settled = false
const done = () => {
if (settled) return
settled = true
const waiters = slot[key]
if (Array.isArray(waiters)) {
const idx = waiters.indexOf(done)
if (idx >= 0) waiters.splice(idx, 1)
}
resolve()
}
slot[key].push(done)
const ms = Math.max(1, Math.min(100, Number(timeoutMs) || 10))
setTimeout(done, ms)
})
if (name === 'socket') {
const fd = st.nextFd++
const k = String(fd)
@@ -6559,6 +6591,8 @@ async function executeKernel(disk, store, swarm, initSource) {
type: sockType,
sockType,
state: 'created',
recvWaiters: [],
acceptWaiters: [],
aliasKeys: new Set([k])
}
this.bareOsRegisterLogicalFd(fd, 'bare-os-socket-bridge:idle')
@@ -6717,6 +6751,7 @@ async function executeKernel(disk, store, swarm, initSource) {
}
: null
})
bridgeNotify(slot, 'recv')
})
this.bareOsLogicalFds[k] =
'bare-os-socket-bridge:udp:' + (host || '127.0.0.1') + ':' + port
@@ -6764,7 +6799,8 @@ async function executeKernel(disk, store, swarm, initSource) {
})
try {
if (connectTimeoutMs > 0) {
let to
/** @type {ReturnType<typeof setTimeout> | null} */
let to = null
const timeoutP = new Promise((_, reject) => {
to = setTimeout(
() =>
@@ -6776,8 +6812,9 @@ async function executeKernel(disk, store, swarm, initSource) {
connectTimeoutMs
)
})
await Promise.race([connectPromise, timeoutP])
clearTimeout(to)
await Promise.race([connectPromise, timeoutP]).finally(() => {
if (to) clearTimeout(to)
})
} else {
await connectPromise
}
@@ -6820,12 +6857,15 @@ async function executeKernel(disk, store, swarm, initSource) {
return
}
slot.tcpRecvQueue.push({ buf: u8.slice() })
bridgeNotify(slot, 'recv')
})
sock.on('end', () => {
slot.tcpPeerEnded = true
bridgeNotify(slot, 'recv')
})
sock.on('error', () => {
slot.tcpSockError = true
bridgeNotify(slot, 'recv')
})
this.bareOsLogicalFds[k] =
'bare-os-socket-bridge:tcp:' + (host || '127.0.0.1') + ':' + port
@@ -6998,6 +7038,7 @@ async function executeKernel(disk, store, swarm, initSource) {
}
: null
})
bridgeNotify(slot, 'recv')
})
this.bareOsLogicalFds[k] =
'bare-os-socket-bridge:udp-bound:' + host + ':' + boundUdpPort
@@ -7125,8 +7166,10 @@ async function executeKernel(disk, store, swarm, initSource) {
return
}
wrap.tcpRecvQueue.push({ buf: u8.slice() })
bridgeNotify(wrap, 'recv')
})
slot.acceptQueue.push(wrap)
bridgeNotify(slot, 'accept')
})
try {
await new Promise((resolve, reject) => {
@@ -7235,6 +7278,7 @@ async function executeKernel(disk, store, swarm, initSource) {
type: 1,
tcpRecvQueue: wrap.tcpRecvQueue,
tcpRecvDropped: wrap.tcpRecvDropped || 0,
recvWaiters: wrap.recvWaiters || [],
aliasKeys: new Set([nk])
}
this.bareOsRegisterLogicalFd(
@@ -7273,7 +7317,7 @@ async function executeKernel(disk, store, swarm, initSource) {
'accept: timed out waiting (same budget as BARE_OS_POSIX_DGRAM_RECV_BLOCK_MS_MAX)'
}
}
await new Promise((r) => setTimeout(r, 10))
await bridgeWait(slot, blockMax - (Date.now() - t0), 'accept')
}
}
if (name === 'shutdown') {
@@ -7550,7 +7594,7 @@ async function executeKernel(disk, store, swarm, initSource) {
tcpRecvDropped: slot.tcpRecvDropped || 0
}
}
await new Promise((r) => setTimeout(r, 10))
await bridgeWait(slot, blockMax - (Date.now() - t0), 'recv')
}
}
if (logicalOp === 'send' || logicalOp === 'sendmsg') {
@@ -7731,7 +7775,7 @@ async function executeKernel(disk, store, swarm, initSource) {
dgramRecvDropped: slot.dgramRecvDropped || 0
}
}
await new Promise((r) => setTimeout(r, 10))
await bridgeWait(slot, blockMax - (Date.now() - t0), 'recv')
}
}
}
packages/bare-os-booter/lib/bare-initd.js
+2 -1
View File
@@ -43,11 +43,12 @@ export const BARE_INITD_UNIT_STATES = Object.freeze([
'starting',
'active',
'stopping',
'skipped',
'failed',
'dead'
])
/** @typedef {{ phase: 'active'|'failed'|'inactive', startedAtMs: number, error?: string }} BareServiceRuntime */
/** @typedef {{ phase: 'active'|'failed'|'inactive'|'starting'|'stopping'|'skipped'|'dead', startedAtMs: number, error?: string }} BareServiceRuntime */
/** @type {BareService[]} */
const registry = []
packages/bare-os-booter/lib/bare-os-bin-offload-worker.cjs
+29 -5
View File
@@ -44,11 +44,17 @@ if (Worker.isMainThread) {
} catch {
/* ignore */
}
finish({ ok: false, reason: 'wasm_time_budget' })
finish({ ok: false, reason: 'wasm_time_budget', error: 'worker terminated by wasm budget' })
}, maxWasmMs)
}
w.on('message', (msg) => finish(msg))
w.on('error', () => finish(null))
w.on('error', (err) =>
finish({
ok: false,
reason: 'worker_error',
error: err && err.message ? err.message : String(err)
})
)
})
}
} else {
@@ -61,11 +67,26 @@ if (Worker.isMainThread) {
const logs = []
/** @type {string[]} */
const errs = []
const MAX_LINES = 2000
let droppedLogs = 0
let droppedErrs = 0
const ctx = {
exitCode: 0,
console: {
log: (...a) => logs.push(a.map(String).join(' ')),
error: (...a) => errs.push(a.map(String).join(' '))
log: (...a) => {
if (logs.length >= MAX_LINES) {
droppedLogs++
return
}
logs.push(a.map(String).join(' '))
},
error: (...a) => {
if (errs.length >= MAX_LINES) {
droppedErrs++
return
}
errs.push(a.map(String).join(' '))
}
}
}
const raw = typeof source === 'string' ? source : ''
@@ -81,11 +102,14 @@ if (Worker.isMainThread) {
exitCode:
ctx.exitCode != null ? Number(ctx.exitCode) || 0 : 0,
logs,
errs
errs,
droppedLogs,
droppedErrs
})
} catch (e) {
Worker.parentPort.postMessage({
ok: false,
reason: 'worker_runtime_error',
error: e && e.message ? e.message : String(e)
})
}
packages/bare-os-booter/lib/bare-os-ipc.js
+13 -3
View File
@@ -99,13 +99,16 @@ class FanoutHub {
* @param {Uint8Array} u8
*/
publish(u8) {
let drops = 0
for (const fifo of this.subscribers) {
try {
fifo.push(u8, this.maxPerSubBytes)
} catch {
/* drop for this subscriber if backlog full */
drops++
}
}
return drops
}
subscribe() {
@@ -152,11 +155,13 @@ export function createBareOsIpc(opts = {}) {
const perChannelMax =
opts.perChannelMaxBytes instanceof Map ? opts.perChannelMaxBytes : null
/** @type {{ fifoCreates: number, fifoCreateDeniedQuota: number, fifoPushDenied: number, mqFull: number }} */
/** @type {{ fifoCreates: number, fifoCreateDeniedQuota: number, fifoPushDenied: number, fifoTakeAborted: number, fanoutDrop: number, mqFull: number }} */
const telemetry = {
fifoCreates: 0,
fifoCreateDeniedQuota: 0,
fifoPushDenied: 0,
fifoTakeAborted: 0,
fanoutDrop: 0,
mqFull: 0
}
@@ -324,7 +329,12 @@ export function createBareOsIpc(opts = {}) {
take(name, takeOpts) {
const ch = channels.get(name)
if (!ch) return Promise.reject(new Error('no such fifo: ' + name))
return ch.take(takeOpts?.signal)
return ch.take(takeOpts?.signal).catch((e) => {
if (e && typeof e === 'object' && /** @type {Error} */ (e).name === 'AbortError') {
telemetry.fifoTakeAborted++
}
throw e
})
},
/**
@@ -384,7 +394,7 @@ export function createBareOsIpc(opts = {}) {
let hub = fanouts.get(name)
if (!hub || hub.subscribers.length === 0) return
const copy = new Uint8Array(u8)
hub.publish(copy)
telemetry.fanoutDrop += Number(hub.publish(copy) || 0)
},
/**
packages/bare-os-booter/lib/bare-os-process-table.js
+24 -1
View File
@@ -2,6 +2,25 @@
* Logical process table for `/proc`-style introspection (guest has no real PIDs).
*/
/** Stable synthetic initd pid mapping for the current booter session. */
const INITD_SYNTH_PID_BASE = 5100
/** @type {Map<string, number>} */
const initdSyntheticPidByName = new Map()
let nextInitdSyntheticPid = INITD_SYNTH_PID_BASE
/**
* @param {string} unitName
*/
function syntheticInitdPidFor(unitName) {
const k = String(unitName || '').trim()
if (!k) return nextInitdSyntheticPid++
const existing = initdSyntheticPidByName.get(k)
if (Number.isFinite(existing)) return existing
const pid = nextInitdSyntheticPid++
initdSyntheticPidByName.set(k, pid)
return pid
}
/**
* @param {Record<string, string | undefined> | null | undefined} env
*/
@@ -288,7 +307,7 @@ export function bareOsProcessTableSnapshot(opts = {}) {
let i = 0
for (const u of initdUnitsRaw) {
if (!u || typeof u.name !== 'string') continue
const pid = 5100 + i
const pid = syntheticInitdPidFor(u.name)
i++
const pgid = pid
const nmLower = u.name.toLowerCase()
@@ -310,6 +329,10 @@ export function bareOsProcessTableSnapshot(opts = {}) {
? 'signaled'
: u.phase === 'starting'
? 'running'
: u.phase === 'stopping'
? 'sleeping'
: u.phase === 'skipped'
? 'sleeping'
: 'sleeping',
startedAtMs: typeof u.startedAtMs === 'number' ? u.startedAtMs : now,
role: 'initd_unit',
packages/bare-os-booter/lib/bare-os-swarm-peer-policy.js
+35 -3
View File
@@ -62,7 +62,9 @@ function readSwarmPolicyEnv(env) {
* banUntilMs: number,
* ewmaLatencyMs: number,
* lastLatencyMs: number | null,
* reconnectBudget: number
* reconnectBudget: number,
* earliestRetryAtMs: number,
* retryTier: 'none' | 'short' | 'medium' | 'long' | 'xlong'
* }} PeerState
*/
@@ -99,13 +101,33 @@ export class BareOsSwarmPeerPolicyEngine {
banUntilMs: 0,
ewmaLatencyMs: 0,
lastLatencyMs: null,
reconnectBudget: 32
reconnectBudget: 32,
earliestRetryAtMs: 0,
retryTier: 'none'
}
this._peers.set(k, st)
}
return st
}
_retryDelayFor(st) {
const n = Math.max(0, Number(st.consecutiveFail) || 0)
const tier =
n >= 9 ? 'xlong' : n >= 6 ? 'long' : n >= 4 ? 'medium' : n >= 2 ? 'short' : 'none'
const base =
tier === 'xlong'
? 15000
: tier === 'long'
? 8000
: tier === 'medium'
? 3000
: tier === 'short'
? 800
: 0
const jitter = base > 0 ? Math.floor(base * 0.3 * Math.random()) : 0
return { tier, delayMs: base + jitter }
}
/**
* @param {string} peerKey
* @param {boolean} ok
@@ -119,6 +141,8 @@ export class BareOsSwarmPeerPolicyEngine {
if (ok) {
st.ok++
st.consecutiveFail = 0
st.retryTier = 'none'
st.earliestRetryAtMs = 0
const lat =
typeof probeMeta.latencyMs === 'number' && Number.isFinite(probeMeta.latencyMs)
? Math.max(0, probeMeta.latencyMs)
@@ -137,6 +161,9 @@ export class BareOsSwarmPeerPolicyEngine {
} else {
st.fail++
st.consecutiveFail++
const retry = this._retryDelayFor(st)
st.retryTier = retry.tier
st.earliestRetryAtMs = now + retry.delayMs
if (st.consecutiveFail >= this._cfg.failsBeforeBan) {
st.banCount++
const base = Math.min(
@@ -146,6 +173,8 @@ export class BareOsSwarmPeerPolicyEngine {
const jitter = Math.floor(base * 0.2 * Math.random())
st.banUntilMs = now + base + jitter
st.consecutiveFail = 0
st.retryTier = 'xlong'
st.earliestRetryAtMs = st.banUntilMs
}
}
}
@@ -174,6 +203,7 @@ export class BareOsSwarmPeerPolicyEngine {
const now = Date.now()
if (!st) return true
if (st.banUntilMs > now) return false
if (st.earliestRetryAtMs > now) return false
if (st.reconnectBudget <= 0) return false
return true
}
@@ -251,7 +281,9 @@ export class BareOsSwarmPeerPolicyEngine {
id,
...st,
banned: st.banUntilMs > now,
banRemainingMs: st.banUntilMs > now ? st.banUntilMs - now : 0
banRemainingMs: st.banUntilMs > now ? st.banUntilMs - now : 0,
retryWaitMs:
st.earliestRetryAtMs > now ? st.earliestRetryAtMs - now : 0
})),
atMs: now
}
packages/bare-os-booter/lib/bare-os-var-log.js
+19
View File
@@ -156,6 +156,25 @@ export async function appendVarLog(ctx, logicalFilePath, kind, line) {
const prev = await vfs.readFile(logicalFilePath)
const ts = new Date().toISOString()
const chunk = ctx.b4a.from(`[${ts}] [${kind}] ${line}\n`)
const fastAppend =
(ctx.env?.BARE_OS_VAR_LOG_FAST_APPEND === '1' ||
ctx.env?.BARE_OS_VAR_LOG_FAST_APPEND === 'true') &&
typeof vfs.appendFile === 'function'
if (fastAppend) {
await vfs.appendFile(logicalFilePath, chunk)
await mirrorBareOsTelemetryNdjson(ctx, {
type: 'varLog',
path: logicalFilePath,
kind,
line: String(line).slice(0, 4000)
})
await mirrorBareOsOtelJsonl(ctx, 'varLog', {
path: logicalFilePath,
kind,
line: String(line).slice(0, 4000)
})
return
}
let merged = prev ? ctx.b4a.concat([prev, chunk]) : chunk
if (merged.length > LOG_MAX_BYTES) {
const start = Math.max(0, merged.length - LOG_KEEP_BYTES)
packages/bare-os-booter/lib/kernel-runner.js
+30 -3
View File
@@ -155,9 +155,25 @@ async function tryRunBinInBareWorker(ctx, argv, source) {
const cmd = argv[0]
const env = ctx.vfs?.env
if (!binWorkerOffloadEnabled(cmd, env)) return false
const reportFallback = (reason, err) => {
if (!env) return
const on =
env.BARE_OS_BIN_WORKER_REPORT_FALLBACK === '1' ||
env.BARE_OS_BIN_WORKER_REPORT_FALLBACK === 'true'
if (!on) return
const msg = err && err.message ? `${reason}: ${err.message}` : String(reason)
try {
ctx.console?.error?.(`[bare-os] bin-worker fallback for ${cmd}: ${msg}`)
} catch {
/* ignore */
}
}
try {
const req = await bareOsCreateRequireFromMetaUrl()
if (!req) return false
if (!req) {
reportFallback('bare-module createRequire unavailable')
return false
}
const { runBinOffloaded } = req('./bare-os-bin-offload-worker.cjs')
const wasmRaw = env && String(env.BARE_OS_BIN_WORKER_WASM_MS_MAX || '').trim()
const wasmN = wasmRaw ? Number.parseInt(wasmRaw, 10) : 0
@@ -169,15 +185,26 @@ async function tryRunBinInBareWorker(ctx, argv, source) {
{ source, argv },
maxWasmMs > 0 ? { maxWasmMs } : undefined
)
if (!msg || msg.ok === false) return false
if (!msg || msg.ok === false) {
reportFallback(
msg && typeof msg === 'object' && msg.reason ? `worker rejected (${msg.reason})` : 'worker rejected'
)
return false
}
const logs = Array.isArray(msg.logs) ? msg.logs : []
const errs = Array.isArray(msg.errs) ? msg.errs : []
for (const line of logs) ctx.console.log(line)
for (const line of errs) ctx.console.error(line)
if (Number(msg.droppedLogs) > 0 || Number(msg.droppedErrs) > 0) {
ctx.console.error(
`[bare-os] bin-worker output truncated for ${cmd} (dropped logs=${Number(msg.droppedLogs) || 0}, errs=${Number(msg.droppedErrs) || 0})`
)
}
ctx.exitCode =
msg.exitCode != null ? Number(msg.exitCode) || 0 : 0
return true
} catch {
} catch (e) {
reportFallback('worker exception', /** @type {Error} */ (e))
return false
}
}
packages/bare-os-booter/lib/shell.js
+18 -1
View File
@@ -1330,7 +1330,10 @@ async function expandShellWordTokens(ctx, wordTok, env, globOpts) {
* @param {boolean} bareOsStdoutCaptured
*/
function bareOsPipelineChildCtx(ctx, env, stdinText, bareOsStdoutCaptured) {
const o = Object.assign({}, ctx, { env, bareOsStdoutCaptured })
const o = Object.assign({}, ctx, {
env: env && typeof env === 'object' ? { ...env } : env,
bareOsStdoutCaptured
})
if (stdinText != null) o.shellStdin = stdinText
return o
}
@@ -2249,6 +2252,20 @@ function scheduleBackgroundShell(ctx, toks) {
.slice(0, 6)
.join(' ')
const childCtx = Object.assign({}, ctx)
if (ctx.env && typeof ctx.env === 'object') {
childCtx.env = { ...ctx.env }
}
if (ctx.vfs && typeof ctx.vfs === 'object' && ctx.vfs.env && typeof ctx.vfs.env === 'object') {
childCtx.vfs = Object.assign(
Object.create(Object.getPrototypeOf(ctx.vfs)),
ctx.vfs,
{ env: childCtx.env && typeof childCtx.env === 'object' ? childCtx.env : { ...ctx.vfs.env } }
)
}
childCtx.shellBackgroundJobs = { nextId: 1, list: [] }
if (ctx.shellSessionState && typeof ctx.shellSessionState === 'object') {
childCtx.shellSessionState = { ...ctx.shellSessionState }
}
/** @type {{ id: number, label: string, promise: Promise<string>, done: boolean, stopped: boolean, pgid: number, sid: number, jobControlModel: string, lastExitCode?: number }} */
const entry = {
id,