Skip to content

Perf: coalesce runtime state snapshots #13

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
MacHatter1 wants to merge 1 commit into
base: main
Choose a base branch
from
+90 −81
Open

Perf: coalesce runtime state snapshots #13

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 52 additions & 81 deletions electron/runtime/sessionManager.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -784,6 +784,7 @@ export class RuntimeSessionManager {
expectedVersion,
events: [],
broadcasts: [],
stateBroadcastQueued: false,
channelCheckpoints: new Map(),
runSessionIdsBefore: new Set(this.#runs.keys()),
deploymentFinalizations: [],
Expand Down Expand Up @@ -834,13 +835,7 @@ export class RuntimeSessionManager {
this.#enqueueSchedulerEvent(event)
this.#queueWorkflowWakeupsForKernelEvent(event)
}
for (const deferred of transaction.broadcasts) {
this.#broadcast(
isObject(deferred) && 'state' in deferred
? { ...deferred, state: this.getState() }
: deferred,
)
}
this.#broadcastDeferredEvents(transaction)
if (
transaction.outboxEffects.length > 0 &&
this.#controlCommandCrashBeforeEffectDrain
Expand Down Expand Up @@ -903,13 +898,7 @@ export class RuntimeSessionManager {
this.#enqueueSchedulerEvent(event)
this.#queueWorkflowWakeupsForKernelEvent(event)
}
for (const deferred of transaction.broadcasts) {
this.#broadcast(
isObject(deferred) && 'state' in deferred
? { ...deferred, state: this.getState() }
: deferred,
)
}
this.#broadcastDeferredEvents(transaction)
queueMicrotask(() => this.#drainWorkflowWakeups())
throw error
}
Expand Down Expand Up @@ -973,6 +962,7 @@ export class RuntimeSessionManager {
actor,
events: [],
broadcasts: [],
stateBroadcastQueued: false,
channelCheckpoints: new Map(),
runSessionIdsBefore: new Set(this.#runs.keys()),
deploymentFinalizations: [],
Expand Down Expand Up @@ -1602,10 +1592,7 @@ export class RuntimeSessionManager {
}
// gate === 'human' (or master with nobody to route to): the slot waits
// for an approve/deny command from the UI/CLI.
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
}

#pendingRequestText(slot, subscription) {
Expand Down Expand Up @@ -1727,10 +1714,7 @@ export class RuntimeSessionManager {
},
)
this.#touch()
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
return { ok: true, slotKey }
}

Expand Down Expand Up @@ -1906,10 +1890,7 @@ export class RuntimeSessionManager {
this.#syncLoopStateForSubscription(subscription, 'activated')
this.#touch()
await this.#stopSubscriptionAtMaxFirings(subscription, ctx)
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
} catch (error) {
console.error(
`Approved activation ${slot.slotKey} failed to execute: ${error instanceof Error ? error.message : String(error)}`,
Expand Down Expand Up @@ -2069,7 +2050,7 @@ export class RuntimeSessionManager {
}, ctx, { reason: group.reason })
this.#touch()
await this.#stopSubscriptionAtMaxFirings(subscription, ctx)
this.#broadcast({ type: 'runtime.state', state: this.getState() })
this.#broadcastState()
return { group: clone(group) }
}

Expand Down Expand Up @@ -2232,10 +2213,7 @@ export class RuntimeSessionManager {
this.#syncLoopStateForSubscription(subscription, 'subscription.authored')
this.#syncTimerForSubscription(subscription)
this.#touch()
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
return {
subscription: clone(subscription),
staticCheck: {
Expand Down Expand Up @@ -2359,10 +2337,7 @@ export class RuntimeSessionManager {
break
}
this.#touch()
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
return { ok: true, subscription: clone(subscription) }
}

Expand Down Expand Up @@ -2700,10 +2675,7 @@ export class RuntimeSessionManager {
)
this.#syncAdapterForSource(source)
this.#touch()
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
return {
source: clone(source),
...(token ? { token } : {}),
Expand Down Expand Up @@ -2750,10 +2722,7 @@ export class RuntimeSessionManager {
delete this.#state.sourceTokens[sourceId]
}
this.#touch()
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
return { ok: true, source: clone(source) }
}

Expand Down Expand Up @@ -4026,10 +3995,7 @@ export class RuntimeSessionManager {
this.#state.sessions[sessionId].archived = archived
this.#appendKernelEvent('session.archived', { sessionId, archived }, ctx)
this.#touch()
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
return { ok: true, state: this.getState() }
}

Expand Down Expand Up @@ -4489,10 +4455,7 @@ export class RuntimeSessionManager {
ctx,
)
this.#touch()
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
return { clusterId, state: this.getState() }
}

Expand Down Expand Up @@ -4523,10 +4486,7 @@ export class RuntimeSessionManager {
if (this.#state.sessions[cluster.masterSessionId]) {
this.#assignMaster(clusterId, cluster.masterSessionId, ctx)
this.#touch()
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
return {
sessionId: cluster.masterSessionId,
state: this.getState(),
Expand Down Expand Up @@ -4561,10 +4521,7 @@ export class RuntimeSessionManager {
)
this.#assignMaster(clusterId, result.sessionId, ctx)
this.#touch()
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
return {
sessionId: result.sessionId,
state: this.getState(),
Expand Down Expand Up @@ -4594,10 +4551,7 @@ export class RuntimeSessionManager {

this.#assignMaster(clusterId, sessionId, ctx)
this.#touch()
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
return { state: this.getState() }
}

Expand Down Expand Up @@ -4626,10 +4580,7 @@ export class RuntimeSessionManager {
ctx,
)
this.#touch()
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
return { state: this.getState() }
}

Expand Down Expand Up @@ -4675,10 +4626,7 @@ export class RuntimeSessionManager {

if (changed) {
this.#touch()
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
}

return { state: this.getState() }
Expand Down Expand Up @@ -5217,7 +5165,7 @@ export class RuntimeSessionManager {
{ reason: optionalTrimmedString(input.reason) },
)
this.#touch()
this.#broadcast({ type: 'runtime.state', state: this.getState() })
this.#broadcastState()
return { ok: true, results, removedDeliveries, removedBytes, state: this.getState() }
}

Expand Down Expand Up @@ -5392,10 +5340,7 @@ export class RuntimeSessionManager {
},
)
this.#touch()
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
}
return { edge: clone(existing) }
}
Expand Down Expand Up @@ -8066,10 +8011,7 @@ export class RuntimeSessionManager {
status: 'running',
})
this.#touch()
this.#broadcast({
type: 'runtime.state',
state: this.getState(),
})
this.#broadcastState()
journalAutomaticDeploymentRunStarted(this.#wf(), sessionId)

let run
Expand Down Expand Up @@ -8989,7 +8931,7 @@ export class RuntimeSessionManager {
this.#releaseWorkspaceLease(runId, 'completed')
this.#runContext.delete(sessionId)
this.#touch()
this.#broadcast({ type: 'runtime.state', state: this.getState() })
this.#broadcastState()
return { ok: true, sessionId, kernelEventId: finishedEvent?.id, state: this.getState() }
}

Expand Down Expand Up @@ -9867,6 +9809,35 @@ export class RuntimeSessionManager {
this.#snapshotPersistTimer.unref?.()
}

#broadcastState() {
const transaction = this.#controlCommandContext.getStore()
if (transaction && transaction.closed !== true) {
if (transaction.stateBroadcastQueued) return
transaction.stateBroadcastQueued = true
transaction.broadcasts.push({ type: 'runtime.state' })
return
}
this.#broadcast({ type: 'runtime.state', state: this.getState() })
}

#broadcastDeferredEvents(transaction: JsonRecord) {
const committedState = transaction.broadcasts.some(
(deferred) =>
isObject(deferred) &&
(deferred.type === 'runtime.state' || 'state' in deferred),
)
? this.getState()
: undefined
for (const deferred of transaction.broadcasts) {
this.#broadcast(
isObject(deferred) &&
(deferred.type === 'runtime.state' || 'state' in deferred)
? { ...deferred, state: committedState }
: deferred,
)
}
}

#broadcast(event) {
const transaction = this.#controlCommandContext.getStore()
if (transaction && transaction.closed !== true) {
Expand Down
38 changes: 38 additions & 0 deletions tests/runtime/runtime-session-manager.test.mjs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -759,6 +759,44 @@ test('compiled RuntimeSessionManager persists provider instance settings', async
}
})

test('runtime state broadcasts coalesce snapshot construction within a command', async () => {
const tempRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'orrery-runtime-state-broadcast-'))
const storageFile = path.join(tempRoot, 'runtime-state.json')
const sessionId = 'snapshot-coalesce-session'
fs.writeFileSync(
storageFile,
JSON.stringify(persistedStateWithSession(sessionId, tempRoot)),
)
const emittedEvents = []
const runtime = new RuntimeSessionManager({
storageFile,
broadcastRuntimeEvent: (event) => emittedEvents.push(event),
})
const originalGetState = runtime.getState.bind(runtime)
let snapshotCalls = 0
runtime.getState = (...args) => {
snapshotCalls += 1
return originalGetState(...args)
}

try {
const result = await runtime.dispatchCommand({
kind: 'archive_session',
actor: { kind: 'human' },
input: { sessionId },
})

assert.equal(snapshotCalls, 2)
const stateEvents = emittedEvents.filter((event) => event.type === 'runtime.state')
assert.equal(stateEvents.length, 1)
assert.equal(stateEvents[0].state.sessions[sessionId].archived, true)
assert.equal(result.state.sessions[sessionId].archived, true)
} finally {
runtime.killAll()
fs.rmSync(tempRoot, { recursive: true, force: true })
}
})

test('old v8 state gains default-grok without rewriting existing sessions', () => {
const tempRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'orrery-grok-v8-normalize-'))
const storageFile = path.join(tempRoot, 'runtime-state.json')
Expand Down