mirror of
https://github.com/stablyai/orca.git
synced 2026-11-15 18:37:24 +00:00
perf(persistence): prepare changed storage leaves before worker saves
This commit is contained in:
@@ -37,6 +37,7 @@ import {
|
||||
import { backfillFolderScopeConnectionIds } from '../../restoring-sessions/folder-scope-migration'
|
||||
import { prepareLoadedTerminalSettings } from './prepare-loaded-terminal-settings'
|
||||
import { prepareLoadedProfileSettings } from './prepare-loaded-profile-settings'
|
||||
import { parseProfileStateInput } from './parse-profile-state-input'
|
||||
import { normalizeLoadedProfileState } from './normalize-loaded-profile-state'
|
||||
|
||||
import type { StoreRuntimeState } from '../store-runtime-state'
|
||||
@@ -55,6 +56,8 @@ type LoadedStateParsingOperationsRuntime = Pick<
|
||||
| 'loadNeedsSave'
|
||||
| 'originalExperimentalNativeChat'
|
||||
| 'protectedSecrets'
|
||||
| 'profileStateAuthority'
|
||||
| 'profileStateSplitSnapshots'
|
||||
| 'storageAuthority'
|
||||
| 'terminalScrollbackSnapshotStorage'
|
||||
>
|
||||
@@ -71,16 +74,23 @@ export class LoadedStateParsingOperations {
|
||||
}
|
||||
|
||||
loadFromAuthority(raw: string | undefined): PersistedState {
|
||||
return this.loadInternal(raw)
|
||||
return this.loadInternal(raw, undefined, true)
|
||||
}
|
||||
|
||||
loadParsedFromAuthority(parsed: Record<string, unknown> | undefined): PersistedState {
|
||||
return this.loadInternal(undefined, parsed)
|
||||
return this.loadInternal(undefined, parsed, true)
|
||||
}
|
||||
|
||||
private loadInternal(serialized?: string, parsedInput?: Record<string, unknown>): PersistedState {
|
||||
private loadInternal(
|
||||
serialized?: string,
|
||||
parsedInput?: Record<string, unknown>,
|
||||
admitted = false
|
||||
): PersistedState {
|
||||
// Pre-telemetry profiles lack cohort fields, so source existence determines upgrade status.
|
||||
const fileExistedOnLoad = serialized !== undefined || parsedInput !== undefined
|
||||
if (admitted && !fileExistedOnLoad && this.runtime.profileStateAuthority?.writePreparedState) {
|
||||
this.runtime.profileStateSplitSnapshots.seed(undefined)
|
||||
}
|
||||
logStartupMilestone('persistence-load-start', {
|
||||
fileExists: fileExistedOnLoad
|
||||
})
|
||||
@@ -90,23 +100,14 @@ export class LoadedStateParsingOperations {
|
||||
let chatUiLaunchTipAudience: ChatUiLaunchTipAudience | null = null
|
||||
try {
|
||||
if (fileExistedOnLoad) {
|
||||
const readStartedAt = performance.now()
|
||||
const raw = parsedInput === undefined ? serialized : undefined
|
||||
if (raw !== undefined) {
|
||||
logStartupMilestone('persistence-read-done', {
|
||||
bytes: Buffer.byteLength(raw),
|
||||
durationMs: Math.round(performance.now() - readStartedAt)
|
||||
})
|
||||
logStartupMilestone('persistence-json-parse-start')
|
||||
parsed = JSON.parse(raw)
|
||||
logStartupMilestone('persistence-json-parse-done')
|
||||
} else {
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Legacy partial records enter the existing domain normalizers through this loader type.
|
||||
parsed = parsedInput as PersistedState
|
||||
}
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Admitted JSON records enter the existing domain normalizers through this loader type.
|
||||
parsed = parseProfileStateInput(serialized, parsedInput) as LoadedProfileState | undefined
|
||||
if (parsed === undefined) {
|
||||
throw new Error('Profile state startup snapshot is missing')
|
||||
}
|
||||
if (admitted && this.runtime.profileStateAuthority?.writePreparedState) {
|
||||
this.runtime.profileStateSplitSnapshots.seed(parsed)
|
||||
}
|
||||
parsed = this.cohorts.migrateCohortSeeds(parsed, true)
|
||||
// Freeze audience before defaults or migrations alter the saved Chat UI state.
|
||||
chatUiLaunchTipAudience = this.cohorts.captureChatUiLaunchTipAudience(parsed)
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
import { logStartupMilestone } from '../../../startup/diagnostics/startup-diagnostics'
|
||||
|
||||
export function parseProfileStateInput(
|
||||
serialized: string | undefined,
|
||||
parsedInput: Record<string, unknown> | undefined
|
||||
): Record<string, unknown> | undefined {
|
||||
const readStartedAt = performance.now()
|
||||
const raw = parsedInput === undefined ? serialized : undefined
|
||||
if (raw === undefined) {
|
||||
return parsedInput
|
||||
}
|
||||
logStartupMilestone('persistence-read-done', {
|
||||
bytes: Buffer.byteLength(raw),
|
||||
durationMs: Math.round(performance.now() - readStartedAt)
|
||||
})
|
||||
logStartupMilestone('persistence-json-parse-start')
|
||||
const parsed = JSON.parse(raw)
|
||||
logStartupMilestone('persistence-json-parse-done')
|
||||
return parsed
|
||||
}
|
||||
@@ -17,6 +17,7 @@ export type PrimaryStateWriteOperationsRuntime = Pick<
|
||||
| 'pendingWrite'
|
||||
| 'profileMaintenancePending'
|
||||
| 'profileStateAuthority'
|
||||
| 'profileStateSplitSnapshots'
|
||||
| 'protectedSecrets'
|
||||
| 'quitFlushStarted'
|
||||
| 'state'
|
||||
@@ -60,6 +61,7 @@ export async function stopAfterFailedPrimaryStateMutation(
|
||||
// Throwing callbacks supply no rollback; freeze persistence rather than commit partial edits.
|
||||
runtime.writesFrozen = true
|
||||
runtime.quitFlushStarted = true
|
||||
runtime.profileStateSplitSnapshots.clear()
|
||||
runtime.fatalMutationError = error instanceof Error ? error : new Error(String(error))
|
||||
if (runtime.writeTimer) {
|
||||
clearTimeout(runtime.writeTimer)
|
||||
@@ -67,3 +69,32 @@ export async function stopAfterFailedPrimaryStateMutation(
|
||||
}
|
||||
await runtime.profileStateAuthority?.close?.()
|
||||
}
|
||||
|
||||
export function capturePrimaryStateWriteIntent(runtime: PrimaryStateWriteOperationsRuntime) {
|
||||
const generation = runtime.writeGeneration
|
||||
const dirtyDomains = runtime.dirtyProfileStateDomains
|
||||
const automationRuns = runtime.pendingAutomationRunsAfter
|
||||
runtime.dirtyProfileStateDomains = new Set()
|
||||
runtime.pendingAutomationRunsAfter = undefined
|
||||
|
||||
const restoreIntent = (): void => {
|
||||
if (dirtyDomains === null) {
|
||||
runtime.dirtyProfileStateDomains = null
|
||||
} else if (runtime.dirtyProfileStateDomains !== null) {
|
||||
for (const domain of dirtyDomains) {
|
||||
runtime.dirtyProfileStateDomains.add(domain)
|
||||
}
|
||||
}
|
||||
if (automationRuns === undefined || runtime.pendingAutomationRunsAfter !== undefined) {
|
||||
return
|
||||
}
|
||||
// A run list replaced while this write was in flight is newer; save it as a whole row instead.
|
||||
if (automationRuns === runtime.state.automationRuns) {
|
||||
runtime.pendingAutomationRunsAfter = automationRuns
|
||||
} else {
|
||||
runtime.dirtyProfileStateDomains?.add('automationRuns')
|
||||
}
|
||||
}
|
||||
|
||||
return { generation, dirtyDomains, automationRuns, restoreIntent }
|
||||
}
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
import { writeSelectiveProfileState } from './profile-state-selective-write'
|
||||
import { profileStateWriterFailureOutcome } from '../../profile-state/writer/profile-state-writer-errors'
|
||||
import type { PrimaryStateWriteOperationsContext } from './primary-state-write-context'
|
||||
import {
|
||||
canReuseDurableProfileState,
|
||||
capturePrimaryStateWriteIntent,
|
||||
markPrimaryStateWriteDurable
|
||||
} from './primary-state-write-runtime'
|
||||
|
||||
@@ -27,6 +29,45 @@ export function writeToDiskSync(
|
||||
opts.expectedGeneration === undefined
|
||||
? undefined
|
||||
: () => runtime.writeGeneration === opts.expectedGeneration
|
||||
if (authority.writePreparedState) {
|
||||
const intent = capturePrimaryStateWriteIntent(runtime)
|
||||
try {
|
||||
const captured = serialization.buildPreparedStateWrite(
|
||||
intent.dirtyDomains,
|
||||
intent.automationRuns
|
||||
)
|
||||
if (isCurrent && !isCurrent()) {
|
||||
intent.restoreIntent()
|
||||
return false
|
||||
}
|
||||
if (captured.stateHash !== null && canReuseDurableProfileState(runtime, captured.stateHash)) {
|
||||
captured.acknowledge()
|
||||
} else {
|
||||
if (captured.write) {
|
||||
authority.writePreparedState(captured.write)
|
||||
} else {
|
||||
authority.writeSerializedState(captured.payload)
|
||||
}
|
||||
captured.acknowledge()
|
||||
runtime.protectedSecrets.commitRetentionUpdates(captured.protectedSecretUpdates)
|
||||
}
|
||||
runtime.lastWrittenStateHash =
|
||||
runtime.writeGeneration === intent.generation ? captured.stateHash : null
|
||||
markPrimaryStateWriteDurable(runtime, intent.generation)
|
||||
} catch (error) {
|
||||
if (profileStateWriterFailureOutcome(error) === 'indeterminate') {
|
||||
runtime.profileStateSplitSnapshots.clear()
|
||||
}
|
||||
intent.restoreIntent()
|
||||
throw error
|
||||
}
|
||||
try {
|
||||
authority.scheduleBackup?.()
|
||||
} catch (error) {
|
||||
console.error('[persistence] Failed to schedule profile state backup:', error)
|
||||
}
|
||||
return true
|
||||
}
|
||||
const selective = writeSelectiveProfileState(
|
||||
authority,
|
||||
serialization,
|
||||
|
||||
@@ -1,6 +1,10 @@
|
||||
import type { AsyncProfileStateAuthority } from './profile-state-authority'
|
||||
import type { PrimaryStateWriteOperationsContext } from './primary-state-write-context'
|
||||
import { markPrimaryStateWriteDurable } from './primary-state-write-runtime'
|
||||
import {
|
||||
capturePrimaryStateWriteIntent,
|
||||
markPrimaryStateWriteDurable
|
||||
} from './primary-state-write-runtime'
|
||||
import { profileStateWriterFailureOutcome } from '../../profile-state/writer/profile-state-writer-errors'
|
||||
import { prepareSelectiveProfileStateWrite } from './profile-state-selective-write'
|
||||
|
||||
// Capture intent now; edits made while this operation waits belong to the next write.
|
||||
@@ -9,30 +13,8 @@ export async function writeProfileStateInWorker(
|
||||
authority: AsyncProfileStateAuthority
|
||||
): Promise<boolean> {
|
||||
authority.assertWritable()
|
||||
const generation = runtime.writeGeneration
|
||||
const dirtyDomains = runtime.dirtyProfileStateDomains
|
||||
const automationRuns = runtime.pendingAutomationRunsAfter
|
||||
runtime.dirtyProfileStateDomains = new Set()
|
||||
runtime.pendingAutomationRunsAfter = undefined
|
||||
|
||||
const restoreIntent = (): void => {
|
||||
if (dirtyDomains === null) {
|
||||
runtime.dirtyProfileStateDomains = null
|
||||
} else if (runtime.dirtyProfileStateDomains !== null) {
|
||||
for (const domain of dirtyDomains) {
|
||||
runtime.dirtyProfileStateDomains.add(domain)
|
||||
}
|
||||
}
|
||||
if (automationRuns === undefined || runtime.pendingAutomationRunsAfter !== undefined) {
|
||||
return
|
||||
}
|
||||
// A run list replaced while this write was in flight is newer; save it as a whole row instead.
|
||||
if (automationRuns === runtime.state.automationRuns) {
|
||||
runtime.pendingAutomationRunsAfter = automationRuns
|
||||
} else {
|
||||
runtime.dirtyProfileStateDomains?.add('automationRuns')
|
||||
}
|
||||
}
|
||||
const { generation, dirtyDomains, automationRuns, restoreIntent } =
|
||||
capturePrimaryStateWriteIntent(runtime)
|
||||
|
||||
try {
|
||||
const prepared = beginWrite(
|
||||
@@ -48,6 +30,7 @@ export async function writeProfileStateInWorker(
|
||||
return false
|
||||
}
|
||||
await prepared.completion
|
||||
prepared.acknowledge?.()
|
||||
runtime.protectedSecrets.commitRetentionUpdates(prepared.protectedSecretUpdates)
|
||||
runtime.lastWrittenStateHash =
|
||||
runtime.writeGeneration === generation ? prepared.stateHash : null
|
||||
@@ -60,6 +43,9 @@ export async function writeProfileStateInWorker(
|
||||
runtime.firstPendingSaveAt = null
|
||||
}
|
||||
} catch (error) {
|
||||
if (profileStateWriterFailureOutcome(error) === 'indeterminate') {
|
||||
runtime.profileStateSplitSnapshots.clear()
|
||||
}
|
||||
restoreIntent()
|
||||
throw error
|
||||
}
|
||||
@@ -80,6 +66,24 @@ function beginWrite(
|
||||
lastWrittenStateHash: string | null,
|
||||
isCurrent: () => boolean
|
||||
) {
|
||||
if (authority.writePreparedState) {
|
||||
const captured = serialization.buildPreparedStateWrite(dirtyDomains, automationRuns)
|
||||
if (!isCurrent()) {
|
||||
return undefined
|
||||
}
|
||||
const unchanged = captured.stateHash !== null && captured.stateHash === lastWrittenStateHash
|
||||
const completion = unchanged
|
||||
? authority.assertCurrentRevision()
|
||||
: captured.write
|
||||
? authority.writePreparedState(captured.write)
|
||||
: authority.writeSerializedState(captured.payload)
|
||||
return {
|
||||
completion,
|
||||
stateHash: captured.stateHash,
|
||||
protectedSecretUpdates: unchanged ? [] : captured.protectedSecretUpdates,
|
||||
acknowledge: captured.acknowledge
|
||||
}
|
||||
}
|
||||
const selective = prepareSelectiveProfileStateWrite(
|
||||
authority,
|
||||
serialization,
|
||||
@@ -94,7 +98,12 @@ function beginWrite(
|
||||
selective.automationRuns !== undefined
|
||||
? authority.writeSerializedAutomationRuns(selective.replacements, selective.automationRuns)
|
||||
: authority.writeSerializedDomains(selective.replacements)
|
||||
return { completion, stateHash: null, protectedSecretUpdates: selective.protectedSecretUpdates }
|
||||
return {
|
||||
completion,
|
||||
stateHash: null,
|
||||
protectedSecretUpdates: selective.protectedSecretUpdates,
|
||||
acknowledge: undefined
|
||||
}
|
||||
}
|
||||
const complete = serialization.buildStateToSave(true)
|
||||
if (!isCurrent()) {
|
||||
@@ -109,6 +118,7 @@ function beginWrite(
|
||||
return {
|
||||
completion,
|
||||
stateHash: complete.stateHash,
|
||||
protectedSecretUpdates: unchanged ? [] : complete.protectedSecretUpdates
|
||||
protectedSecretUpdates: unchanged ? [] : complete.protectedSecretUpdates,
|
||||
acknowledge: undefined
|
||||
}
|
||||
}
|
||||
|
||||
+6
-3
@@ -54,7 +54,10 @@ export function serializeCompleteProfileStateDomains(
|
||||
}
|
||||
}
|
||||
|
||||
function serializeSelectiveProfileStateDomainFragment(domain: string, value: unknown): string {
|
||||
export function serializeSelectiveProfileStateDomainFragment(
|
||||
domain: string,
|
||||
value: unknown
|
||||
): string {
|
||||
let needsNormalization = false
|
||||
const fragment = serializeProfileStateDomainFragment(domain, value, (_key, entry) => {
|
||||
if (entry !== null && typeof entry === 'object') {
|
||||
@@ -70,7 +73,7 @@ function serializeSelectiveProfileStateDomainFragment(domain: string, value: unk
|
||||
return needsNormalization ? JSON.stringify(JSON.parse(fragment.toWellFormed())) : fragment
|
||||
}
|
||||
|
||||
function serializeProfileStateDomainFragment(
|
||||
export function serializeProfileStateDomainFragment(
|
||||
domain: string,
|
||||
value: unknown,
|
||||
replacer?: (key: string, value: unknown) => unknown
|
||||
@@ -79,6 +82,6 @@ function serializeProfileStateDomainFragment(
|
||||
return JSON.stringify({ [domain]: value }, replacer)
|
||||
}
|
||||
|
||||
function extractProfileStateDomainPayload(domain: string, fragment: string): string {
|
||||
export function extractProfileStateDomainPayload(domain: string, fragment: string): string {
|
||||
return fragment.slice(JSON.stringify(domain).length + 2, -1)
|
||||
}
|
||||
|
||||
@@ -1,6 +1,14 @@
|
||||
import type { LegacyPaneKeyAliasEntry } from '../../../../shared/persisted-state-types'
|
||||
import type { AutomationRun } from '../../../../shared/automations/history/automation-run'
|
||||
import type { ProfileStateDatabaseQuarantine } from '../../profile-state/recovery/profile-state-database-quarantine'
|
||||
import type { ProfileStateDomainPatch } from '../../profile-state/documents/profile-state-domain-patches'
|
||||
|
||||
export type PreparedProfileStateWrite = {
|
||||
replacements: readonly ProfileStateDomainReplacement[]
|
||||
patches: readonly ProfileStateDomainPatch[]
|
||||
runPayloads?: readonly string[]
|
||||
complete?: boolean
|
||||
}
|
||||
|
||||
export type ProfileStateMaintenance = {
|
||||
resume(): Promise<void>
|
||||
@@ -17,6 +25,8 @@ export type ProfileStateAuthority = {
|
||||
/** Without selective domain writes, callers replace the full document. */
|
||||
writeSerializedDomains?: (replacements: readonly ProfileStateDomainReplacement[]) => void
|
||||
|
||||
writePreparedState?: (write: PreparedProfileStateWrite) => void
|
||||
|
||||
/** Commit definitions and run projections atomically without serializing unrelated domains. */
|
||||
writeSerializedAutomationRuns?: (
|
||||
replacements: readonly ProfileStateDomainReplacement[],
|
||||
@@ -48,6 +58,7 @@ export type AsyncProfileStateAuthority = Omit<
|
||||
| 'asynchronous'
|
||||
| 'assertCurrentRevision'
|
||||
| 'writeSerializedDomains'
|
||||
| 'writePreparedState'
|
||||
| 'writeSerializedAutomationRuns'
|
||||
| 'writeSerializedState'
|
||||
| 'writeCompleteSerializedDomains'
|
||||
@@ -60,6 +71,7 @@ export type AsyncProfileStateAuthority = Omit<
|
||||
abort(): Promise<void>
|
||||
assertCurrentRevision(): Promise<void>
|
||||
writeSerializedDomains(replacements: readonly ProfileStateDomainReplacement[]): Promise<void>
|
||||
writePreparedState?: (write: PreparedProfileStateWrite) => Promise<void>
|
||||
writeSerializedAutomationRuns(
|
||||
replacements: readonly ProfileStateDomainReplacement[],
|
||||
runs: readonly AutomationRun[]
|
||||
|
||||
+5
@@ -8,6 +8,7 @@ import { ProfileStateSqliteAuthority } from '../../profile-state/profile-state-s
|
||||
import { writeVersionedProfileStateExport } from '../../profile-state/legacy-json/profile-state-versioned-export'
|
||||
import type {
|
||||
AsyncProfileStateAuthority,
|
||||
PreparedProfileStateWrite,
|
||||
ProfileStateDomainReplacement
|
||||
} from './profile-state-authority'
|
||||
import { Store } from '../store'
|
||||
@@ -58,6 +59,10 @@ export class DelayedAuthority implements AsyncProfileStateAuthority {
|
||||
const captured = Buffer.from(payload)
|
||||
await this.dispatch(() => this.inner.writeSerializedState(captured), true)
|
||||
}
|
||||
async writePreparedState(write: PreparedProfileStateWrite) {
|
||||
const captured = structuredClone(write)
|
||||
await this.dispatch(() => this.inner.writePreparedState(captured), true)
|
||||
}
|
||||
async writeSerializedDomains(replacements: readonly ProfileStateDomainReplacement[]) {
|
||||
const captured = structuredClone(replacements)
|
||||
this.captures.push([...captured])
|
||||
|
||||
@@ -26,6 +26,7 @@ export function freezeProfileStateWrites(runtime: StoreRuntimeState): void {
|
||||
runtime.writeTimer = null
|
||||
}
|
||||
suspendGithubCachePersistence(runtime)
|
||||
runtime.profileStateSplitSnapshots.clear()
|
||||
runtime.profileStateAuthority?.close?.()
|
||||
}
|
||||
|
||||
@@ -44,6 +45,7 @@ export async function freezeProfileStateWritesAsync(runtime: StoreRuntimeState):
|
||||
])
|
||||
} finally {
|
||||
runtime.writesFrozen = true
|
||||
runtime.profileStateSplitSnapshots.clear()
|
||||
await runtime.profileStateAuthority?.close?.()
|
||||
}
|
||||
}
|
||||
@@ -150,6 +152,7 @@ async function pauseProfileState(
|
||||
}
|
||||
signal?.throwIfAborted()
|
||||
runtime.writesFrozen = true
|
||||
runtime.profileStateSplitSnapshots.clear()
|
||||
if (!flush) {
|
||||
await authority?.close?.()
|
||||
return {
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
import type { AutomationRun } from '../../../../shared/automations/history/automation-run'
|
||||
import type { ProtectedSecretRetentionUpdate } from '../../../credentials/protected-secret-persistence'
|
||||
import { PROFILE_STATE_SPLIT_DOMAIN_DEPTHS } from '../../profile-state/document-parts/profile-state-document-parts'
|
||||
import type { ProfileStateDomainPatch } from '../../profile-state/documents/profile-state-domain-patches'
|
||||
import type { ProfileStateSplitSnapshots } from './profile-state-split-snapshots'
|
||||
import type {
|
||||
ProfileStateDomainReplacement,
|
||||
PreparedProfileStateWrite
|
||||
} from './profile-state-authority'
|
||||
import {
|
||||
serializeProfileStateDomainFragment,
|
||||
serializeSelectiveProfileStateDomainFragment,
|
||||
extractProfileStateDomainPayload
|
||||
} from './profile-state-authority-writes'
|
||||
import {
|
||||
applySecretSentinelSubstitutions,
|
||||
type SecretSentinelSubstitution
|
||||
} from './secret-sentinel-substitution'
|
||||
|
||||
export type CapturedProfileStateWrite = {
|
||||
protectedSecretUpdates: ProtectedSecretRetentionUpdate[]
|
||||
acknowledge: () => void
|
||||
} & (
|
||||
| { write: PreparedProfileStateWrite; stateHash: null; payload?: never }
|
||||
| { write?: never; stateHash: string; payload: Buffer }
|
||||
)
|
||||
|
||||
type PreparedWriteProjection = {
|
||||
snapshots: ProfileStateSplitSnapshots
|
||||
stateToSave: Record<string, unknown>
|
||||
complete: boolean
|
||||
secretSubs: readonly SecretSentinelSubstitution[]
|
||||
protectedSecretUpdates: ProtectedSecretRetentionUpdate[]
|
||||
automationRuns?: readonly AutomationRun[]
|
||||
serializeRoot?: () => {
|
||||
payload: Buffer
|
||||
stateHash: string
|
||||
protectedSecretUpdates: ProtectedSecretRetentionUpdate[]
|
||||
}
|
||||
}
|
||||
|
||||
export function capturePreparedProfileStateWrite({
|
||||
snapshots,
|
||||
stateToSave,
|
||||
complete,
|
||||
secretSubs,
|
||||
protectedSecretUpdates,
|
||||
automationRuns,
|
||||
serializeRoot
|
||||
}: PreparedWriteProjection): CapturedProfileStateWrite {
|
||||
const candidates = new Map<string, unknown>()
|
||||
const acknowledge = snapshots.acknowledge.bind(snapshots, candidates)
|
||||
if (complete && 'toJSON' in stateToSave && typeof stateToSave.toJSON === 'function') {
|
||||
const built = serializeRoot!()
|
||||
const state: unknown = JSON.parse(built.payload.toString('utf8'))
|
||||
for (const domain of PROFILE_STATE_SPLIT_DOMAIN_DEPTHS.keys()) {
|
||||
candidates.set(
|
||||
domain,
|
||||
state !== null && typeof state === 'object'
|
||||
? Object.getOwnPropertyDescriptor(state, domain)?.value
|
||||
: undefined
|
||||
)
|
||||
}
|
||||
return { ...built, acknowledge }
|
||||
}
|
||||
const replacements: ProfileStateDomainReplacement[] = []
|
||||
const patches: ProfileStateDomainPatch[] = []
|
||||
// Complete serialization captures entries before any hook; selective reads each key in turn.
|
||||
function* selectiveEntries(): Generator<[string, unknown]> {
|
||||
for (const domain of Object.keys(stateToSave)) {
|
||||
yield [domain, stateToSave[domain]]
|
||||
}
|
||||
}
|
||||
for (const [domain, value] of complete ? Object.entries(stateToSave) : selectiveEntries()) {
|
||||
const split = PROFILE_STATE_SPLIT_DOMAIN_DEPTHS.has(domain)
|
||||
const captured = split ? snapshots.capture(domain, value) : undefined
|
||||
if (captured) {
|
||||
patches.push(captured.patch)
|
||||
candidates.set(domain, captured.value)
|
||||
continue
|
||||
}
|
||||
const fragment = complete
|
||||
? serializeProfileStateDomainFragment(domain, value)
|
||||
: serializeSelectiveProfileStateDomainFragment(domain, value)
|
||||
const serialized =
|
||||
complete && fragment !== '{}'
|
||||
? applySecretSentinelSubstitutions(fragment, secretSubs, '', 'text').payload
|
||||
: fragment
|
||||
const payload =
|
||||
serialized === '{}' ? null : extractProfileStateDomainPayload(domain, serialized)
|
||||
if (!complete || split || payload !== null) {
|
||||
replacements.push({ domain, payload })
|
||||
}
|
||||
if (split) {
|
||||
candidates.set(domain, payload === null ? undefined : JSON.parse(payload))
|
||||
}
|
||||
}
|
||||
if (complete) {
|
||||
for (const domain of PROFILE_STATE_SPLIT_DOMAIN_DEPTHS.keys()) {
|
||||
if (!candidates.has(domain)) {
|
||||
candidates.set(domain, undefined)
|
||||
}
|
||||
}
|
||||
}
|
||||
const runPayloads =
|
||||
!complete && automationRuns !== undefined
|
||||
? automationRuns.map((run) => {
|
||||
const payload = JSON.stringify(run)
|
||||
if (payload === undefined) {
|
||||
throw new Error('Automation run is not serializable')
|
||||
}
|
||||
return payload
|
||||
})
|
||||
: undefined
|
||||
return {
|
||||
write: { replacements, patches, complete, runPayloads },
|
||||
stateHash: null,
|
||||
protectedSecretUpdates,
|
||||
acknowledge
|
||||
}
|
||||
}
|
||||
+118
@@ -0,0 +1,118 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { ProfileStateSplitSnapshots } from './profile-state-split-snapshots'
|
||||
import {
|
||||
applyProfileStateDomainPatch,
|
||||
prepareProfileStateDomainPatches
|
||||
} from '../../profile-state/documents/profile-state-domain-patches'
|
||||
|
||||
describe('profile state split capture', () => {
|
||||
it('keeps storage and acknowledged captures independent of later in-place edits', () => {
|
||||
const state = { workspaceSession: { tabsByWorktree: { a: [{ title: 'saved' }] } } }
|
||||
const snapshots = new ProfileStateSplitSnapshots()
|
||||
snapshots.seed(state)
|
||||
state.workspaceSession.tabsByWorktree.a[0].title = 'C'
|
||||
const captured = snapshots.capture('workspaceSession', state.workspaceSession)!
|
||||
expect(captured.patch.operations).toEqual([
|
||||
{ kind: 'set', path: ['tabsByWorktree', 'a'], payload: '[{"title":"C"}]' }
|
||||
])
|
||||
state.workspaceSession.tabsByWorktree.a[0].title = 'D'
|
||||
// The first capture failed: retry still compares against the admitted storage value.
|
||||
expect(snapshots.capture('workspaceSession', state.workspaceSession)!.patch.operations).toEqual(
|
||||
[{ kind: 'set', path: ['tabsByWorktree', 'a'], payload: '[{"title":"D"}]' }]
|
||||
)
|
||||
snapshots.acknowledge(new Map([['workspaceSession', captured.value]]))
|
||||
expect(snapshots.capture('workspaceSession', state.workspaceSession)!.patch.operations).toEqual(
|
||||
[{ kind: 'set', path: ['tabsByWorktree', 'a'], payload: '[{"title":"D"}]' }]
|
||||
)
|
||||
state.workspaceSession.tabsByWorktree.a[0].title = 'C'
|
||||
expect(snapshots.capture('workspaceSession', state.workspaceSession)!.patch.operations).toEqual(
|
||||
[]
|
||||
)
|
||||
})
|
||||
|
||||
it('preserves native JSON meaning, own keys and order through logical patches', () => {
|
||||
const snapshots = new ProfileStateSplitSnapshots()
|
||||
snapshots.seed({ workspaceSession: { tabsByWorktree: { a: [], b: [] }, keep: null } })
|
||||
const current: unknown = JSON.parse(
|
||||
'{"tabsByWorktree":{"b":[],"__proto__":[{"title":"own"}],"a":[]},"keep":{}}'
|
||||
)
|
||||
const captured = snapshots.capture('workspaceSession', current)!
|
||||
const [prepared] = prepareProfileStateDomainPatches([captured.patch])
|
||||
const result = applyProfileStateDomainPatch(
|
||||
{ tabsByWorktree: { a: [], b: [] }, keep: null },
|
||||
prepared
|
||||
)
|
||||
expect(JSON.stringify(result)).toBe(
|
||||
'{"tabsByWorktree":{"b":[],"__proto__":[{"title":"own"}],"a":[]},"keep":{}}'
|
||||
)
|
||||
snapshots.acknowledge(new Map([['workspaceSession', captured.value]]))
|
||||
expect(
|
||||
snapshots.capture('workspaceSession', {
|
||||
...(current as Record<string, unknown>),
|
||||
optional: undefined
|
||||
})!.patch.operations
|
||||
).toEqual([])
|
||||
})
|
||||
|
||||
it('normalizes array undefined and nonfinite numbers without changing JSON values', () => {
|
||||
const snapshots = new ProfileStateSplitSnapshots()
|
||||
snapshots.seed({ workspaceSession: { a: { b: [null, null, 0] } } })
|
||||
expect(
|
||||
snapshots.capture('workspaceSession', {
|
||||
a: { b: [undefined, Number.NaN, -0] },
|
||||
optional: undefined
|
||||
})!.patch.operations
|
||||
).toEqual([])
|
||||
})
|
||||
|
||||
it('leaves hooks, proxies and sparse arrays to native serialization without observing them', () => {
|
||||
if (!('rawJSON' in JSON) || typeof JSON.rawJSON !== 'function') {
|
||||
throw new Error('Raw JSON fixture requires Node 24 or newer')
|
||||
}
|
||||
const snapshots = new ProfileStateSplitSnapshots()
|
||||
snapshots.seed({ workspaceSession: {} })
|
||||
let observations = 0
|
||||
const getter = {
|
||||
get title() {
|
||||
observations++
|
||||
return 'native'
|
||||
}
|
||||
}
|
||||
const hook = {
|
||||
toJSON() {
|
||||
observations++
|
||||
return 'native'
|
||||
}
|
||||
}
|
||||
const proxy = new Proxy(
|
||||
{},
|
||||
{
|
||||
ownKeys() {
|
||||
observations++
|
||||
return []
|
||||
}
|
||||
}
|
||||
)
|
||||
for (const value of [getter, hook, proxy, Array(1), JSON.rawJSON('42')]) {
|
||||
expect(snapshots.capture('workspaceSession', { a: { b: value } })).toBeUndefined()
|
||||
}
|
||||
expect(observations).toBe(0)
|
||||
})
|
||||
|
||||
it.each([100, 1000])('sends only the changed leaf with %i workspace rows', (count) => {
|
||||
const tabsByWorktree = Object.fromEntries(
|
||||
Array.from({ length: count }, (_, index) => [`w${index}`, [{ title: 'unchanged' }]])
|
||||
)
|
||||
const snapshots = new ProfileStateSplitSnapshots()
|
||||
snapshots.seed({ workspaceSession: { tabsByWorktree } })
|
||||
tabsByWorktree.w0[0].title = 'changed'
|
||||
const captured = snapshots.capture('workspaceSession', { tabsByWorktree })!
|
||||
// One title update needs one small leaf; unrelated row payloads and full key vectors are unnecessary.
|
||||
expect(captured.patch).toEqual({
|
||||
domain: 'workspaceSession',
|
||||
operations: [
|
||||
{ kind: 'set', path: ['tabsByWorktree', 'w0'], payload: '[{"title":"changed"}]' }
|
||||
]
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,231 @@
|
||||
import { types } from 'node:util'
|
||||
import { PROFILE_STATE_SPLIT_DOMAIN_DEPTHS } from '../../profile-state/document-parts/profile-state-document-parts'
|
||||
import {
|
||||
defineProfileStateOwnValue,
|
||||
type ProfileStateDomainPatch,
|
||||
type ProfileStateDomainPatchOperation
|
||||
} from '../../profile-state/documents/profile-state-domain-patches'
|
||||
|
||||
const unsupported = Symbol('native domain serialization required')
|
||||
const lookupGetter = Object.prototype.__lookupGetter__
|
||||
const lookupSetter = Object.prototype.__lookupSetter__
|
||||
|
||||
function record(value: unknown): value is Record<string, unknown> {
|
||||
return value !== null && typeof value === 'object' && !Array.isArray(value)
|
||||
}
|
||||
|
||||
function admitObject(value: object): void {
|
||||
if (types.isProxy(value)) {
|
||||
throw unsupported
|
||||
}
|
||||
const prototype: unknown = Object.getPrototypeOf(value)
|
||||
if (
|
||||
Array.isArray(value)
|
||||
? prototype !== Array.prototype
|
||||
: prototype !== Object.prototype && prototype !== null
|
||||
) {
|
||||
throw unsupported
|
||||
}
|
||||
admitToJson(value)
|
||||
if ('isRawJSON' in JSON && typeof JSON.isRawJSON === 'function' && JSON.isRawJSON(value)) {
|
||||
throw unsupported
|
||||
}
|
||||
}
|
||||
|
||||
function admitToJson(value: object): void {
|
||||
if (Object.hasOwn(value, 'toJSON')) {
|
||||
const descriptor = Object.getOwnPropertyDescriptor(value, 'toJSON')!
|
||||
if (!('value' in descriptor) || typeof descriptor.value === 'function') {
|
||||
throw unsupported
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function admitPrototypes(): void {
|
||||
if (
|
||||
Object.getPrototypeOf(Object.prototype) !== null ||
|
||||
Object.getPrototypeOf(Array.prototype) !== Object.prototype
|
||||
) {
|
||||
throw unsupported
|
||||
}
|
||||
admitToJson(Object.prototype)
|
||||
admitToJson(Array.prototype)
|
||||
}
|
||||
|
||||
function readDataProperty(value: object, key: string): unknown {
|
||||
if (lookupGetter.call(value, key) || lookupSetter.call(value, key)) {
|
||||
throw unsupported
|
||||
}
|
||||
const entry: unknown = Reflect.get(value, key)
|
||||
// Empty accessors also read undefined; ordinary optional undefined data remains eligible.
|
||||
if (entry === undefined && !('value' in Object.getOwnPropertyDescriptor(value, key)!)) {
|
||||
throw unsupported
|
||||
}
|
||||
return entry
|
||||
}
|
||||
|
||||
function enter(value: object, ancestors: Set<object>): void {
|
||||
admitObject(value)
|
||||
if (ancestors.has(value)) {
|
||||
throw unsupported
|
||||
}
|
||||
ancestors.add(value)
|
||||
}
|
||||
|
||||
/** Complete admission even after a mismatch; native stringify must never discover a hidden hook. */
|
||||
function equalJson(value: unknown, previous: unknown, ancestors: Set<object>): boolean {
|
||||
if (value === null || value === undefined) {
|
||||
return previous === null
|
||||
}
|
||||
switch (typeof value) {
|
||||
case 'string':
|
||||
case 'boolean':
|
||||
return value === previous
|
||||
case 'number':
|
||||
return (Number.isFinite(value) ? value : null) === previous
|
||||
case 'object':
|
||||
break
|
||||
default:
|
||||
throw unsupported
|
||||
}
|
||||
enter(value, ancestors)
|
||||
try {
|
||||
if (Array.isArray(value)) {
|
||||
const old = Array.isArray(previous) ? previous : []
|
||||
let equal = Array.isArray(previous) && old.length === value.length
|
||||
for (let index = 0; index < value.length; index++) {
|
||||
if (!Object.hasOwn(value, index)) {
|
||||
throw unsupported
|
||||
}
|
||||
const childEqual = equalJson(readDataProperty(value, String(index)), old[index], ancestors)
|
||||
equal &&= childEqual
|
||||
}
|
||||
return equal
|
||||
}
|
||||
const old = record(previous) ? previous : undefined
|
||||
const oldKeys = old ? Object.keys(old) : []
|
||||
let equal = old !== undefined
|
||||
let index = 0
|
||||
for (const key of Object.keys(value)) {
|
||||
const child = readDataProperty(value, key)
|
||||
if (child === undefined) {
|
||||
continue
|
||||
}
|
||||
const childEqual = equalJson(child, old?.[key], ancestors)
|
||||
equal &&= key === oldKeys[index] && childEqual
|
||||
index++
|
||||
}
|
||||
return equal && index === oldKeys.length
|
||||
} finally {
|
||||
ancestors.delete(value)
|
||||
}
|
||||
}
|
||||
|
||||
export type CapturedProfileStateSplitDomain = {
|
||||
patch: ProfileStateDomainPatch
|
||||
value: unknown
|
||||
}
|
||||
|
||||
/** Only admitted storage or a successful commit may replace these independent JSON trees. */
|
||||
export class ProfileStateSplitSnapshots {
|
||||
private readonly acknowledged = new Map<string, unknown>()
|
||||
|
||||
seed(state: Record<string, unknown> | undefined): void {
|
||||
this.acknowledged.clear()
|
||||
for (const domain of PROFILE_STATE_SPLIT_DOMAIN_DEPTHS.keys()) {
|
||||
this.acknowledged.set(domain, structuredClone(state?.[domain]))
|
||||
}
|
||||
}
|
||||
|
||||
clear(): void {
|
||||
this.acknowledged.clear()
|
||||
}
|
||||
|
||||
acknowledge(candidates: ReadonlyMap<string, unknown>): void {
|
||||
for (const [domain, value] of candidates) {
|
||||
this.acknowledged.set(domain, value)
|
||||
}
|
||||
}
|
||||
|
||||
capture(domain: string, value: unknown): CapturedProfileStateSplitDomain | undefined {
|
||||
const depth = PROFILE_STATE_SPLIT_DOMAIN_DEPTHS.get(domain)
|
||||
if (depth === undefined || !this.acknowledged.has(domain) || value === undefined) {
|
||||
return undefined
|
||||
}
|
||||
const operations: ProfileStateDomainPatchOperation[] = []
|
||||
const path: string[] = []
|
||||
const ancestors = new Set<object>()
|
||||
const set = (current: unknown): unknown => {
|
||||
const payload = JSON.stringify(current)
|
||||
if (payload === undefined) {
|
||||
throw unsupported
|
||||
}
|
||||
operations.push({ kind: 'set', path: [...path], payload })
|
||||
return JSON.parse(payload)
|
||||
}
|
||||
const visit = (current: unknown, previous: unknown, remaining: number): unknown => {
|
||||
if (remaining === 0 || !record(current) || !record(previous)) {
|
||||
return equalJson(current, previous, ancestors) ? previous : set(current)
|
||||
}
|
||||
enter(current, ancestors)
|
||||
try {
|
||||
const oldKeys = Object.keys(previous)
|
||||
const keys: string[] = []
|
||||
let next = previous
|
||||
const change = (): Record<string, unknown> => {
|
||||
if (next === previous) {
|
||||
next = { ...previous }
|
||||
}
|
||||
return next
|
||||
}
|
||||
for (const key of Object.keys(current)) {
|
||||
const child = readDataProperty(current, key)
|
||||
if (child === undefined) {
|
||||
continue
|
||||
}
|
||||
keys.push(key)
|
||||
path.push(key)
|
||||
const captured = visit(
|
||||
child,
|
||||
Object.hasOwn(previous, key) ? previous[key] : undefined,
|
||||
remaining - 1
|
||||
)
|
||||
path.pop()
|
||||
if (!Object.hasOwn(previous, key) || captured !== previous[key]) {
|
||||
defineProfileStateOwnValue(change(), key, captured)
|
||||
}
|
||||
}
|
||||
const present = new Set(keys)
|
||||
for (const key of oldKeys) {
|
||||
if (!present.has(key)) {
|
||||
operations.push({ kind: 'remove', path: [...path, key] })
|
||||
delete change()[key]
|
||||
}
|
||||
}
|
||||
const naturalKeys = next === previous ? oldKeys : Object.keys(next)
|
||||
if (naturalKeys.some((key, index) => key !== keys[index])) {
|
||||
const ordered: Record<string, unknown> = {}
|
||||
for (const key of keys) {
|
||||
defineProfileStateOwnValue(ordered, key, next[key])
|
||||
}
|
||||
operations.push({ kind: 'order', path: [...path], keys })
|
||||
next = ordered
|
||||
}
|
||||
return next
|
||||
} finally {
|
||||
ancestors.delete(current)
|
||||
}
|
||||
}
|
||||
try {
|
||||
// Eligible capture invokes no user code, so prototype admission holds for this traversal only.
|
||||
admitPrototypes()
|
||||
const candidate = visit(value, this.acknowledged.get(domain), depth)
|
||||
return { patch: { domain, operations }, value: candidate }
|
||||
} catch (error) {
|
||||
if (error !== unsupported) {
|
||||
throw error
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
}
|
||||
+28
@@ -3,6 +3,34 @@ import { fixture } from './profile-state-delayed-authority-fixture'
|
||||
|
||||
export function registerProfileStateWorkerCoordinationTests(): void {
|
||||
describe('worker-owned Store writes', () => {
|
||||
it.each([false, true])(
|
||||
'saves in-place edits made during a split-domain write (failure=%s)',
|
||||
async (fail) => {
|
||||
const { store, authority, readState } = await fixture()
|
||||
vi.spyOn(console, 'error').mockImplementation(() => {})
|
||||
const session = store.getWorkspaceSession()
|
||||
session.activeTabId = 'C'
|
||||
store.setWorkspaceSession(session)
|
||||
const gate = authority.pause()
|
||||
const first = store.flushPendingOrThrowAsync({ drainToStableGeneration: false })
|
||||
const settled = fail ? expect(first).rejects.toThrow('disk refused') : first
|
||||
await gate.started.promise
|
||||
session.activeTabId = 'D'
|
||||
store.setWorkspaceSession(session)
|
||||
if (fail) {
|
||||
gate.finish.reject(new Error('disk refused'))
|
||||
} else {
|
||||
gate.finish.resolve()
|
||||
}
|
||||
await settled
|
||||
if (!fail) {
|
||||
expect(readState().workspaceSession.activeTabId).toBe('C')
|
||||
}
|
||||
await store.flushPendingOrThrowAsync()
|
||||
expect(readState().workspaceSession.activeTabId).toBe('D')
|
||||
}
|
||||
)
|
||||
|
||||
it('retains a newer edit after an older write is acknowledged', async () => {
|
||||
const { store, authority, readState } = await fixture()
|
||||
const gate = authority.pause()
|
||||
|
||||
+1
-3
@@ -54,9 +54,7 @@ export function registerProfileStateWriteBatchingTests(): void {
|
||||
store.flushPendingOrThrowAsync({ drainToStableGeneration: false }),
|
||||
store.flushPendingOrThrowAsync({ drainToStableGeneration: false })
|
||||
].map((waiter) => expect(waiter).rejects.toThrow('disk refused'))
|
||||
vi.spyOn(authority, 'writeCompleteSerializedDomains').mockRejectedValueOnce(
|
||||
new Error('disk refused')
|
||||
)
|
||||
vi.spyOn(authority, 'writePreparedState').mockRejectedValueOnce(new Error('disk refused'))
|
||||
gate.finish.resolve()
|
||||
await Promise.all([first, ...failed])
|
||||
await store.flushPendingOrThrowAsync({ drainToStableGeneration: false })
|
||||
|
||||
+93
-24
@@ -23,6 +23,12 @@ import {
|
||||
type SecretSentinelSubstitution
|
||||
} from './secret-sentinel-substitution'
|
||||
import type { StoreRuntimeState } from '../store-runtime-state'
|
||||
import type { AutomationRun } from '../../../../shared/automations/history/automation-run'
|
||||
import { defineProfileStateOwnValue } from '../../profile-state/documents/profile-state-domain-patches'
|
||||
import {
|
||||
capturePreparedProfileStateWrite,
|
||||
type CapturedProfileStateWrite
|
||||
} from './profile-state-prepared-write'
|
||||
|
||||
/** Domains whose saved form differs from state; whole saves keep this order after the durable spread. */
|
||||
const PROJECTED_SAVE_DOMAINS = [
|
||||
@@ -62,12 +68,54 @@ function withDependentSaveDomains(domains: ReadonlySet<string>): ReadonlySet<str
|
||||
|
||||
type StateSerializationSecretHandlingOperationsRuntime = Pick<
|
||||
StoreRuntimeState,
|
||||
'protectedSecrets' | 'state' | 'originalExperimentalNativeChat'
|
||||
'protectedSecrets' | 'state' | 'originalExperimentalNativeChat' | 'profileStateSplitSnapshots'
|
||||
>
|
||||
|
||||
export class StateSerializationSecretHandlingOperations {
|
||||
constructor(private readonly runtime: StateSerializationSecretHandlingOperationsRuntime) {}
|
||||
|
||||
buildPreparedStateWrite(
|
||||
dirtyDomains: ReadonlySet<string> | null,
|
||||
automationRuns?: readonly AutomationRun[]
|
||||
): CapturedProfileStateWrite {
|
||||
const complete =
|
||||
dirtyDomains === null ||
|
||||
dirtyDomains.size === 0 ||
|
||||
this.runtime.protectedSecrets.hasPendingEncryption()
|
||||
let stateToSave: Record<string, unknown>
|
||||
let secretSubs: SecretSentinelSubstitution[] = []
|
||||
let protectedSecretUpdates: ProtectedSecretRetentionUpdate[] = []
|
||||
const projection = complete ? this.buildCompleteProjection() : undefined
|
||||
if (projection) {
|
||||
;({ stateToSave, secretSubs, protectedSecretUpdates } = projection)
|
||||
} else {
|
||||
stateToSave = {}
|
||||
const encrypt = (slot: string, plaintext: string): string => {
|
||||
const encrypted = this.runtime.protectedSecrets.encrypt(slot, plaintext)
|
||||
if (encrypted.retentionUpdate) {
|
||||
protectedSecretUpdates.push(encrypted.retentionUpdate)
|
||||
}
|
||||
return encrypted.blob
|
||||
}
|
||||
for (const domain of withDependentSaveDomains(dirtyDomains ?? new Set())) {
|
||||
if (domain !== 'automationRuns' || automationRuns === undefined) {
|
||||
defineProfileStateOwnValue(stateToSave, domain, this.projectDomainToSave(domain, encrypt))
|
||||
}
|
||||
}
|
||||
}
|
||||
return capturePreparedProfileStateWrite({
|
||||
snapshots: this.runtime.profileStateSplitSnapshots,
|
||||
stateToSave,
|
||||
complete,
|
||||
secretSubs,
|
||||
protectedSecretUpdates,
|
||||
automationRuns,
|
||||
serializeRoot: projection
|
||||
? () => this.serializeCompleteProjection(projection, false)
|
||||
: undefined
|
||||
})
|
||||
}
|
||||
|
||||
getDurableState(): Omit<PersistedState, 'githubCache'> {
|
||||
const { githubCache: _memoryOnly, ...durable } = this.runtime.state
|
||||
return durable
|
||||
@@ -109,29 +157,23 @@ export class StateSerializationSecretHandlingOperations {
|
||||
stateHash: string
|
||||
protectedSecretUpdates: ProtectedSecretRetentionUpdate[]
|
||||
} {
|
||||
// Mint slot UUIDs after user data is captured; ciphertext searches could match user data and hide real changes.
|
||||
const secretSubs: SecretSentinelSubstitution[] = []
|
||||
const protectedSecretUpdates: ProtectedSecretRetentionUpdate[] = []
|
||||
let protectedStorageDegraded = false
|
||||
const encryptToSentinel = (slot: string, plaintext: string): string => {
|
||||
const encrypted = this.runtime.protectedSecrets.encrypt(slot, plaintext)
|
||||
if (encrypted.retentionUpdate) {
|
||||
protectedSecretUpdates.push(encrypted.retentionUpdate)
|
||||
}
|
||||
protectedStorageDegraded ||= encrypted.degraded
|
||||
const { blob, hashValue = plaintext } = encrypted
|
||||
if (blob === plaintext && hashValue === plaintext) {
|
||||
return blob
|
||||
}
|
||||
const sentinel = `orca-secret-slot-${randomUUID()}`
|
||||
secretSubs.push({ sentinel, blob, hashValue })
|
||||
return sentinel
|
||||
}
|
||||
// Clone before encryption to keep in-memory state plaintext.
|
||||
const stateToSave: Record<string, unknown> = { ...this.getDurableState() }
|
||||
for (const domain of PROJECTED_SAVE_DOMAINS) {
|
||||
stateToSave[domain] = this.projectDomainToSave(domain, encryptToSentinel)
|
||||
}
|
||||
return this.serializeCompleteProjection(this.buildCompleteProjection(), serializeDomains)
|
||||
}
|
||||
|
||||
private serializeCompleteProjection(
|
||||
{
|
||||
stateToSave,
|
||||
secretSubs,
|
||||
protectedStorageDegraded,
|
||||
protectedSecretUpdates
|
||||
}: ReturnType<StateSerializationSecretHandlingOperations['buildCompleteProjection']>,
|
||||
serializeDomains: boolean
|
||||
): {
|
||||
domains?: readonly ProfileStateDomainReplacement[]
|
||||
payload: Buffer
|
||||
stateHash: string
|
||||
protectedSecretUpdates: ProtectedSecretRetentionUpdate[]
|
||||
} {
|
||||
if (
|
||||
serializeDomains &&
|
||||
!('toJSON' in stateToSave && typeof stateToSave.toJSON === 'function')
|
||||
@@ -160,6 +202,33 @@ export class StateSerializationSecretHandlingOperations {
|
||||
return { payload, stateHash, protectedSecretUpdates }
|
||||
}
|
||||
|
||||
private buildCompleteProjection() {
|
||||
// Mint slot UUIDs after user data is captured; ciphertext searches could match user data and hide real changes.
|
||||
const secretSubs: SecretSentinelSubstitution[] = []
|
||||
const protectedSecretUpdates: ProtectedSecretRetentionUpdate[] = []
|
||||
let protectedStorageDegraded = false
|
||||
const encryptToSentinel = (slot: string, plaintext: string): string => {
|
||||
const encrypted = this.runtime.protectedSecrets.encrypt(slot, plaintext)
|
||||
if (encrypted.retentionUpdate) {
|
||||
protectedSecretUpdates.push(encrypted.retentionUpdate)
|
||||
}
|
||||
protectedStorageDegraded ||= encrypted.degraded
|
||||
const { blob, hashValue = plaintext } = encrypted
|
||||
if (blob === plaintext && hashValue === plaintext) {
|
||||
return blob
|
||||
}
|
||||
const sentinel = `orca-secret-slot-${randomUUID()}`
|
||||
secretSubs.push({ sentinel, blob, hashValue })
|
||||
return sentinel
|
||||
}
|
||||
// Clone before encryption to keep in-memory state plaintext.
|
||||
const stateToSave: Record<string, unknown> = { ...this.getDurableState() }
|
||||
for (const domain of PROJECTED_SAVE_DOMAINS) {
|
||||
stateToSave[domain] = this.projectDomainToSave(domain, encryptToSentinel)
|
||||
}
|
||||
return { stateToSave, secretSubs, protectedStorageDegraded, protectedSecretUpdates }
|
||||
}
|
||||
|
||||
/** The one saved form of each domain, shared by selective and whole saves. */
|
||||
private projectDomainToSave(
|
||||
domain: string,
|
||||
|
||||
@@ -35,6 +35,7 @@ import type { ProfileStatePersistenceAuthority } from './profile-writing/profile
|
||||
import type { AutomationRun } from '../../../shared/automations/history/automation-run'
|
||||
import { resolveNewPaneShellPath } from '../migrating-orcad-catalog/terminal-history/orcad-shell-history-installer'
|
||||
import { OrcadTerminalRestoreHintStore } from '../migrating-orcad-catalog/terminal-history/orcad-terminal-restore-hint-store'
|
||||
import { ProfileStateSplitSnapshots } from './profile-writing/profile-state-split-snapshots'
|
||||
|
||||
export type DurableProfileStateMutation<T> = {
|
||||
value: T
|
||||
@@ -79,6 +80,7 @@ export class StoreRuntimeState {
|
||||
quitFlushStarted = false
|
||||
quitFlushPromise: Promise<void> | null = null
|
||||
lastWrittenStateHash: string | null = null
|
||||
readonly profileStateSplitSnapshots = new ProfileStateSplitSnapshots()
|
||||
lastDurableWriteGeneration = -1
|
||||
/** Required durable generation for locally committed migrations; disk-loaded receipts have no entry. */
|
||||
readonly orcadMigrationCommitWriteGenerations = new Map<string, number>()
|
||||
|
||||
+16
-5
@@ -28,6 +28,11 @@ type PartsMeta = {
|
||||
legacyRevision: number | null
|
||||
}
|
||||
|
||||
export type ProfileStateExistingDocumentParts = Map<
|
||||
string,
|
||||
{ ordinal: number; contentHash: string }
|
||||
>
|
||||
|
||||
function readPartsMeta(db: Database.Database, domain: string): PartsMeta | undefined {
|
||||
if (!hasProfileStateDocumentPartsTables(db)) {
|
||||
return undefined
|
||||
@@ -99,7 +104,8 @@ export function authoritativeMeta(
|
||||
export function readProfileStateSplitDocument(
|
||||
db: Database.Database,
|
||||
domain: string,
|
||||
legacyRevision: number
|
||||
legacyRevision: number,
|
||||
retainedParts?: ProfileStateExistingDocumentParts
|
||||
): ProfileStateDocument | undefined {
|
||||
if (!isProfileStateSplitDomain(domain)) {
|
||||
return undefined
|
||||
@@ -110,7 +116,7 @@ export function readProfileStateSplitDocument(
|
||||
}
|
||||
const rows = db
|
||||
.prepare(
|
||||
`SELECT part_key, payload, content_hash FROM ${PROFILE_STATE_DOCUMENT_PARTS_TABLE}
|
||||
`SELECT part_key, ordinal, payload, content_hash FROM ${PROFILE_STATE_DOCUMENT_PARTS_TABLE}
|
||||
WHERE domain = ? ORDER BY ordinal`
|
||||
)
|
||||
.iterate(domain)
|
||||
@@ -128,6 +134,10 @@ export function readProfileStateSplitDocument(
|
||||
)
|
||||
}
|
||||
parts.push({ key: row.part_key, payload: row.payload })
|
||||
retainedParts?.set(row.part_key, {
|
||||
ordinal: Number(row.ordinal),
|
||||
contentHash: String(row.content_hash)
|
||||
})
|
||||
}
|
||||
let payload: string
|
||||
try {
|
||||
@@ -208,6 +218,7 @@ export function writeProfileStateDomainParts(
|
||||
legacyRevision: number
|
||||
/** Existing parts are a valid base to diff against. */
|
||||
diffExisting: boolean
|
||||
existingParts?: ProfileStateExistingDocumentParts
|
||||
}
|
||||
): boolean {
|
||||
if (!hasProfileStateDocumentPartsTables(db)) {
|
||||
@@ -226,8 +237,8 @@ export function writeProfileStateDomainParts(
|
||||
deleteProfileStateDomainParts(db, args.domain)
|
||||
return false
|
||||
}
|
||||
const existing = new Map<string, { ordinal: number; contentHash: string }>()
|
||||
if (args.diffExisting) {
|
||||
const existing = args.existingParts ?? new Map<string, { ordinal: number; contentHash: string }>()
|
||||
if (args.diffExisting && args.existingParts === undefined) {
|
||||
for (const row of db
|
||||
.prepare(
|
||||
`SELECT part_key, ordinal, content_hash FROM ${PROFILE_STATE_DOCUMENT_PARTS_TABLE}
|
||||
@@ -241,7 +252,7 @@ export function writeProfileStateDomainParts(
|
||||
})
|
||||
}
|
||||
}
|
||||
} else {
|
||||
} else if (!args.diffExisting) {
|
||||
db.prepare(`DELETE FROM ${PROFILE_STATE_DOCUMENT_PARTS_TABLE} WHERE domain = ?`).run(
|
||||
args.domain
|
||||
)
|
||||
|
||||
+104
@@ -70,6 +70,110 @@ const v2 =
|
||||
'{"runtime:a":{"tabsByWorktree":{"wt2":[{"id":"t2"}],"wt3":[],"wt1":[{"id":"t1","title":"renamed"}]},"activeTabId":"t1"},"ssh:b":{"tabsByWorktree":{}}}'
|
||||
|
||||
describe('profile state document parts', () => {
|
||||
it('applies sparse patches, complete omissions and empty documents with a shared revision fence', () => {
|
||||
const db = open()
|
||||
try {
|
||||
const initial = writeProfileStateDomains(db, {
|
||||
expectedRevision: 0,
|
||||
replacements: [],
|
||||
complete: true,
|
||||
patches: [
|
||||
{
|
||||
domain: 'workspaceSession',
|
||||
operations: [{ kind: 'set', path: [], payload: '{"tabsByWorktree":{"a":[],"b":[]}}' }]
|
||||
}
|
||||
]
|
||||
})
|
||||
expect(initial.revision).toBe(1)
|
||||
write(db, { workspaceSession: '{"tabsByWorktree":{"a":[],"b":[]}}', obsolete: '1' })
|
||||
const revision = readProfileStateRevision(db)
|
||||
const result = writeProfileStateDomains(db, {
|
||||
expectedRevision: revision,
|
||||
replacements: [],
|
||||
complete: true,
|
||||
patches: [{ domain: 'workspaceSession', operations: [] }]
|
||||
})
|
||||
expect(result.revision).toBe(revision + 1)
|
||||
expect(exportProfileStateJson(db)).toBe(
|
||||
'{"workspaceSession":{"tabsByWorktree":{"a":[],"b":[]}}}'
|
||||
)
|
||||
expect(() =>
|
||||
writeProfileStateDomains(db, {
|
||||
expectedRevision: revision,
|
||||
replacements: [],
|
||||
patches: [{ domain: 'workspaceSession', operations: [] }]
|
||||
})
|
||||
).toThrow('revision changed')
|
||||
writeProfileStateDomains(db, {
|
||||
expectedRevision: result.revision,
|
||||
replacements: [],
|
||||
patches: [
|
||||
{ domain: 'workspaceSession', operations: [{ kind: 'set', path: [], payload: '{}' }] }
|
||||
]
|
||||
})
|
||||
expect(exportProfileStateJson(db)).toBe('{"workspaceSession":{}}')
|
||||
writeProfileStateDomains(db, {
|
||||
expectedRevision: readProfileStateRevision(db),
|
||||
replacements: [],
|
||||
complete: true
|
||||
})
|
||||
expect(exportProfileStateJson(db)).toBe('{}')
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('normalizes an unchanged legacy document through an empty patch', () => {
|
||||
const db = open()
|
||||
try {
|
||||
write(db, { workspaceSession: '{ "tabsByWorktree": { "a": [] } }' })
|
||||
writeProfileStateDomains(db, {
|
||||
expectedRevision: readProfileStateRevision(db),
|
||||
replacements: [],
|
||||
patches: [{ domain: 'workspaceSession', operations: [] }]
|
||||
})
|
||||
expect(exportProfileStateJson(db)).toBe('{"workspaceSession":{"tabsByWorktree":{"a":[]}}}')
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects a changed patch when an untouched part no longer matches the assembled hash', () => {
|
||||
const db = open()
|
||||
try {
|
||||
write(db, { workspaceSession: '{"tabsByWorktree":{"a":[],"b":[]}}' })
|
||||
db.prepare(`UPDATE profile_state_document_parts SET payload = ?, content_hash = ?
|
||||
WHERE domain = 'workspaceSession' AND part_key = ?`).run(
|
||||
'["corrupt"]',
|
||||
hashProfileStatePayload('["corrupt"]'),
|
||||
'["tabsByWorktree","b"]'
|
||||
)
|
||||
const revision = readProfileStateRevision(db)
|
||||
expect(
|
||||
writeProfileStateDomains(db, {
|
||||
expectedRevision: revision,
|
||||
replacements: [],
|
||||
patches: [{ domain: 'workspaceSession', operations: [] }]
|
||||
})
|
||||
).toMatchObject({ changed: false, revision })
|
||||
expect(() =>
|
||||
writeProfileStateDomains(db, {
|
||||
expectedRevision: revision,
|
||||
replacements: [],
|
||||
patches: [
|
||||
{
|
||||
domain: 'workspaceSession',
|
||||
operations: [{ kind: 'set', path: ['tabsByWorktree', 'a'], payload: '["changed"]' }]
|
||||
}
|
||||
]
|
||||
})
|
||||
).toThrow('parts hash mismatch')
|
||||
expect(readProfileStateRevision(db)).toBe(revision)
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('reads back the exact document after entries change, appear and move', () => {
|
||||
const db = open()
|
||||
try {
|
||||
|
||||
@@ -3,10 +3,11 @@ import type { openProfileStateDatabase } from '../database/profile-state-databas
|
||||
|
||||
export function buildCompleteDocumentReplacements(
|
||||
db: ReturnType<typeof openProfileStateDatabase>['db'],
|
||||
replacements: readonly ProfileStateDomainReplacement[]
|
||||
replacements: readonly ProfileStateDomainReplacement[],
|
||||
retainedDomains: readonly string[] = []
|
||||
): ProfileStateDomainReplacement[] {
|
||||
const domains = new Set(replacements.map(({ domain }) => domain))
|
||||
const incoming = new Set(domains)
|
||||
const incoming = new Set([...domains, ...retainedDomains])
|
||||
for (const row of db
|
||||
.prepare(`SELECT domain FROM profile_state_documents
|
||||
UNION SELECT domain FROM profile_state_automation_runs_meta WHERE presence <> 'document'`)
|
||||
|
||||
@@ -81,52 +81,72 @@ export function importProfileStateJson(
|
||||
if (options.expectedRevision !== undefined && actualRevision !== options.expectedRevision) {
|
||||
throw new ProfileStateRevisionConflictError(options.expectedRevision, actualRevision)
|
||||
}
|
||||
const revision = actualRevision + 1
|
||||
clearProfileStateAutomationRuns(db)
|
||||
// Imported rows are authoritative; the next write of a split domain re-splits it.
|
||||
clearProfileStateDocumentParts(db)
|
||||
db.exec('DELETE FROM profile_state_documents')
|
||||
const automationRuns = entries.find((entry) => entry.domain === 'automationRuns')
|
||||
const normalizedRuns =
|
||||
automationRuns !== undefined &&
|
||||
rebuildProfileStateAutomationRunsProjection(
|
||||
db,
|
||||
automationRuns.payload,
|
||||
PROFILE_STATE_DOCUMENT_VERSION,
|
||||
updatedAt,
|
||||
revision
|
||||
)
|
||||
const insert = db.prepare(
|
||||
`INSERT INTO profile_state_documents
|
||||
(domain, payload, domain_version, revision, updated_at, content_hash)
|
||||
VALUES (?, ?, ?, ?, ?, ?)`
|
||||
return replaceProfileStateDomainsInTransaction(
|
||||
db,
|
||||
entries,
|
||||
actualRevision,
|
||||
updatedAt,
|
||||
options.acceptedLegacyJsonHash
|
||||
)
|
||||
for (const entry of entries) {
|
||||
const payload = normalizedRuns && entry.domain === 'automationRuns' ? 'null' : entry.payload
|
||||
insert.run(
|
||||
entry.domain,
|
||||
payload,
|
||||
PROFILE_STATE_DOCUMENT_VERSION,
|
||||
revision,
|
||||
updatedAt,
|
||||
hashProfileStatePayload(payload)
|
||||
)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/** Reuse import application inside an already fenced prepared write, without nesting transactions. */
|
||||
export function replaceProfileStateDomainsInTransaction(
|
||||
db: Database.Database,
|
||||
entries: readonly { domain: string; payload: string }[],
|
||||
previousRevision: number,
|
||||
updatedAt: number,
|
||||
acceptedLegacyJsonHash?: string
|
||||
): number {
|
||||
if (!db.isTransaction) {
|
||||
throw new Error('Profile state replacement requires an active write transaction')
|
||||
}
|
||||
const revision = previousRevision + 1
|
||||
clearProfileStateAutomationRuns(db)
|
||||
// Imported rows are authoritative; the next write of a split domain re-splits it.
|
||||
clearProfileStateDocumentParts(db)
|
||||
db.exec('DELETE FROM profile_state_documents')
|
||||
const automationRuns = entries.find((entry) => entry.domain === 'automationRuns')
|
||||
const normalizedRuns =
|
||||
automationRuns !== undefined &&
|
||||
rebuildProfileStateAutomationRunsProjection(
|
||||
db,
|
||||
automationRuns.payload,
|
||||
PROFILE_STATE_DOCUMENT_VERSION,
|
||||
updatedAt,
|
||||
revision
|
||||
)
|
||||
const insert = db.prepare(
|
||||
`INSERT INTO profile_state_documents
|
||||
(domain, payload, domain_version, revision, updated_at, content_hash)
|
||||
VALUES (?, ?, ?, ?, ?, ?)`
|
||||
)
|
||||
for (const entry of entries) {
|
||||
const payload = normalizedRuns && entry.domain === 'automationRuns' ? 'null' : entry.payload
|
||||
insert.run(
|
||||
entry.domain,
|
||||
payload,
|
||||
PROFILE_STATE_DOCUMENT_VERSION,
|
||||
revision,
|
||||
updatedAt,
|
||||
hashProfileStatePayload(payload)
|
||||
)
|
||||
}
|
||||
db.prepare(
|
||||
`INSERT INTO profile_state_meta (key, value) VALUES (?, ?)
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value`
|
||||
).run(PROFILE_STATE_META_REVISION, String(revision))
|
||||
if (acceptedLegacyJsonHash !== undefined) {
|
||||
db.prepare(
|
||||
`INSERT INTO profile_state_meta (key, value) VALUES (?, ?)
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value`
|
||||
).run(PROFILE_STATE_META_REVISION, String(revision))
|
||||
if (options.acceptedLegacyJsonHash !== undefined) {
|
||||
db.prepare(
|
||||
`INSERT INTO profile_state_meta (key, value) VALUES (?, ?)
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value`
|
||||
).run(
|
||||
PROFILE_STATE_META_LEGACY_JSON_ACCEPTANCE,
|
||||
JSON.stringify({ jsonHash: options.acceptedLegacyJsonHash, acceptedRevision: revision })
|
||||
)
|
||||
}
|
||||
return revision
|
||||
})
|
||||
).run(
|
||||
PROFILE_STATE_META_LEGACY_JSON_ACCEPTANCE,
|
||||
JSON.stringify({ jsonHash: acceptedLegacyJsonHash, acceptedRevision: revision })
|
||||
)
|
||||
}
|
||||
return revision
|
||||
}
|
||||
|
||||
export function profileStateJsonMatchesAcceptance(db: Database.Database, rawJson: string): boolean {
|
||||
|
||||
@@ -0,0 +1,164 @@
|
||||
import { PROFILE_STATE_SPLIT_DOMAIN_DEPTHS } from '../document-parts/profile-state-document-parts'
|
||||
import { isRecord } from './profile-state-document-validation'
|
||||
|
||||
export type ProfileStateDomainPatchOperation =
|
||||
| { kind: 'set'; path: readonly string[]; payload: string }
|
||||
| { kind: 'remove'; path: readonly string[] }
|
||||
| { kind: 'order'; path: readonly string[]; keys: readonly string[] }
|
||||
|
||||
export type ProfileStateDomainPatch = {
|
||||
domain: string
|
||||
operations: readonly ProfileStateDomainPatchOperation[]
|
||||
}
|
||||
|
||||
type PreparedOperation =
|
||||
| { kind: 'set'; path: readonly string[]; value: unknown }
|
||||
| Exclude<ProfileStateDomainPatchOperation, { kind: 'set' }>
|
||||
|
||||
export type PreparedProfileStateDomainPatch = {
|
||||
domain: string
|
||||
operations: readonly PreparedOperation[]
|
||||
}
|
||||
|
||||
type PathNode = { mutation?: boolean; order?: boolean; children: Map<string, PathNode> }
|
||||
|
||||
export function prepareProfileStateDomainPatches(
|
||||
patches: readonly ProfileStateDomainPatch[]
|
||||
): PreparedProfileStateDomainPatch[] {
|
||||
if (!Array.isArray(patches)) {
|
||||
throw new Error('Profile state patches must be an array')
|
||||
}
|
||||
const domains = new Set<string>()
|
||||
return patches.map((patch) => {
|
||||
const depth = isRecord(patch) ? PROFILE_STATE_SPLIT_DOMAIN_DEPTHS.get(patch.domain) : undefined
|
||||
if (depth === undefined || domains.has(patch.domain) || !Array.isArray(patch.operations)) {
|
||||
throw new Error('Profile state patch domain is invalid or repeated')
|
||||
}
|
||||
domains.add(patch.domain)
|
||||
const root: PathNode = { children: new Map() }
|
||||
const operations = patch.operations.map((operation): PreparedOperation => {
|
||||
if (
|
||||
!isRecord(operation) ||
|
||||
!Array.isArray(operation.path) ||
|
||||
operation.path.length > depth ||
|
||||
!operation.path.every((key: unknown) => typeof key === 'string') ||
|
||||
!['set', 'remove', 'order'].includes(operation.kind)
|
||||
) {
|
||||
throw new Error('Profile state patch operation is invalid')
|
||||
}
|
||||
let node = root
|
||||
for (const key of operation.path) {
|
||||
if (node.mutation || node.order) {
|
||||
throw new Error('Profile state patch operations conflict')
|
||||
}
|
||||
let child = node.children.get(key)
|
||||
if (!child) {
|
||||
child = { children: new Map() }
|
||||
node.children.set(key, child)
|
||||
}
|
||||
node = child
|
||||
}
|
||||
if (node.mutation || node.order) {
|
||||
throw new Error('Profile state patch operations repeat a path')
|
||||
}
|
||||
if (operation.kind === 'order') {
|
||||
if (
|
||||
operation.path.length >= depth ||
|
||||
!Array.isArray(operation.keys) ||
|
||||
!operation.keys.every((key: unknown) => typeof key === 'string') ||
|
||||
new Set(operation.keys).size !== operation.keys.length
|
||||
) {
|
||||
throw new Error('Profile state patch order is invalid')
|
||||
}
|
||||
node.order = true
|
||||
return { kind: 'order', path: operation.path, keys: operation.keys }
|
||||
}
|
||||
if (node.children.size > 0 || (operation.kind === 'remove' && operation.path.length === 0)) {
|
||||
throw new Error('Profile state patch mutations conflict')
|
||||
}
|
||||
node.mutation = true
|
||||
if (operation.kind === 'remove') {
|
||||
return { kind: 'remove', path: operation.path }
|
||||
}
|
||||
if (typeof operation.payload !== 'string') {
|
||||
throw new Error('Profile state patch payload is invalid')
|
||||
}
|
||||
let value: unknown
|
||||
try {
|
||||
value = JSON.parse(operation.payload)
|
||||
} catch {
|
||||
throw new Error('Profile state patch payload is invalid JSON')
|
||||
}
|
||||
return { kind: 'set', path: operation.path, value }
|
||||
})
|
||||
return { domain: patch.domain, operations }
|
||||
})
|
||||
}
|
||||
|
||||
export function defineProfileStateOwnValue(
|
||||
object: Record<string, unknown>,
|
||||
key: string,
|
||||
value: unknown
|
||||
): void {
|
||||
Object.defineProperty(object, key, {
|
||||
value,
|
||||
enumerable: true,
|
||||
configurable: true,
|
||||
writable: true
|
||||
})
|
||||
}
|
||||
|
||||
/** The worker owns this parsed base; no live Store references reach it. */
|
||||
export function applyProfileStateDomainPatch(
|
||||
base: unknown,
|
||||
patch: PreparedProfileStateDomainPatch
|
||||
): unknown {
|
||||
let value = base
|
||||
for (const operation of patch.operations) {
|
||||
let parent: unknown = value
|
||||
const parentPath = operation.path.slice(0, -1)
|
||||
for (const key of parentPath) {
|
||||
if (!isRecord(parent) || !Object.hasOwn(parent, key)) {
|
||||
throw new Error('Profile state patch parent is missing')
|
||||
}
|
||||
parent = parent[key]
|
||||
}
|
||||
const key = operation.path.at(-1)
|
||||
if (key !== undefined && !isRecord(parent)) {
|
||||
throw new Error('Profile state patch parent is not an object')
|
||||
}
|
||||
const replace = (next: unknown): void => {
|
||||
if (key === undefined) {
|
||||
value = next
|
||||
} else if (isRecord(parent)) {
|
||||
defineProfileStateOwnValue(parent, key, next)
|
||||
}
|
||||
}
|
||||
if (operation.kind === 'set') {
|
||||
replace(operation.value)
|
||||
} else if (operation.kind === 'remove') {
|
||||
if (!isRecord(parent) || key === undefined || !Object.hasOwn(parent, key)) {
|
||||
throw new Error('Profile state patch removal is missing')
|
||||
}
|
||||
delete parent[key]
|
||||
} else {
|
||||
const current = key === undefined ? value : isRecord(parent) ? parent[key] : undefined
|
||||
if (
|
||||
!isRecord(current) ||
|
||||
Object.keys(current).length !== operation.keys.length ||
|
||||
!operation.keys.every((entry) => Object.hasOwn(current, entry))
|
||||
) {
|
||||
throw new Error('Profile state patch order does not cover its object')
|
||||
}
|
||||
const ordered: Record<string, unknown> = {}
|
||||
for (const entry of operation.keys) {
|
||||
defineProfileStateOwnValue(ordered, entry, current[entry])
|
||||
}
|
||||
if (Object.keys(ordered).some((entry, index) => entry !== operation.keys[index])) {
|
||||
throw new Error('Profile state patch order is not native JSON order')
|
||||
}
|
||||
replace(ordered)
|
||||
}
|
||||
}
|
||||
return value
|
||||
}
|
||||
+41
-7
@@ -1,12 +1,17 @@
|
||||
import { PROFILE_STATE_DOCUMENT_VERSION } from '../database/profile-state-database-schema'
|
||||
import { isProfileStateSplitDomain } from '../document-parts/profile-state-document-parts'
|
||||
import './profile-state-documents'
|
||||
import { hashProfileStatePayload as hashProfileStateJson } from './profile-state-document-validation'
|
||||
import {
|
||||
hashProfileStatePayload as hashProfileStateJson,
|
||||
parseProfileStateRoot
|
||||
} from './profile-state-document-validation'
|
||||
import type { AutomationRunsWritePreparation } from '../automation-runs/profile-state-automation-runs-writer'
|
||||
import type {
|
||||
ProfileStateDomainMutation,
|
||||
ProfileStateDomainTransaction
|
||||
} from './profile-state-domain-writes'
|
||||
import { prepareProfileStateDomainPatches } from './profile-state-domain-patches'
|
||||
import type { ProfileStateExistingDocumentParts } from '../document-parts/profile-state-document-parts-storage'
|
||||
|
||||
export type PreparedProfileStateMutation = ProfileStateDomainMutation & {
|
||||
domainVersion: number
|
||||
@@ -15,18 +20,20 @@ export type PreparedProfileStateMutation = ProfileStateDomainMutation & {
|
||||
automationRunsValue?: unknown
|
||||
/** Parsed payload of a split domain, so storing its parts does not parse it again. */
|
||||
splitDomainValue?: unknown
|
||||
existingParts?: ProfileStateExistingDocumentParts
|
||||
automationRuns?: AutomationRunsWritePreparation
|
||||
}
|
||||
|
||||
export function validateProfileStateDomainTransaction(
|
||||
transaction: ProfileStateDomainTransaction
|
||||
): void {
|
||||
export function validateProfileStateDomainTransaction(transaction: ProfileStateDomainTransaction) {
|
||||
if (!Number.isSafeInteger(transaction.expectedRevision) || transaction.expectedRevision < 0) {
|
||||
throw new Error('Profile state expected revision is invalid')
|
||||
}
|
||||
if (
|
||||
!Array.isArray(transaction.replacements) ||
|
||||
(transaction.replacements.length === 0 && transaction.automationRunsAfter === undefined)
|
||||
(transaction.replacements.length === 0 &&
|
||||
transaction.automationRunsAfter === undefined &&
|
||||
transaction.patches === undefined &&
|
||||
!transaction.complete)
|
||||
) {
|
||||
throw new Error('Profile state domain transaction requires at least one replacement')
|
||||
}
|
||||
@@ -41,17 +48,28 @@ export function validateProfileStateDomainTransaction(
|
||||
if (transaction.automationRunsAfter !== undefined && domains.has('automationRuns')) {
|
||||
throw new Error('Profile state automationRuns delta repeats domain: automationRuns')
|
||||
}
|
||||
if (transaction.complete !== undefined && typeof transaction.complete !== 'boolean') {
|
||||
throw new Error('Profile state complete intent is invalid')
|
||||
}
|
||||
const patches = prepareProfileStateDomainPatches(transaction.patches ?? [])
|
||||
for (const patch of patches) {
|
||||
if (domains.has(patch.domain)) {
|
||||
throw new Error(`Profile state patch repeats replacement: ${patch.domain}`)
|
||||
}
|
||||
}
|
||||
return patches
|
||||
}
|
||||
|
||||
export function prepareProfileStateDomainMutation(
|
||||
replacement: ProfileStateDomainMutation
|
||||
replacement: ProfileStateDomainMutation,
|
||||
validatedValue?: { value: unknown }
|
||||
): PreparedProfileStateMutation {
|
||||
const payloadHash =
|
||||
replacement.payload === null ? null : hashProfileStateJson(replacement.payload)
|
||||
let parsed: unknown
|
||||
if (replacement.payload !== null) {
|
||||
try {
|
||||
parsed = JSON.parse(replacement.payload)
|
||||
parsed = validatedValue ? validatedValue.value : JSON.parse(replacement.payload)
|
||||
} catch {
|
||||
throw new Error(`Profile state domain payload is invalid JSON: ${replacement.domain}`)
|
||||
}
|
||||
@@ -81,3 +99,19 @@ function validateProfileStateDomainMutation(replacement: ProfileStateDomainMutat
|
||||
throw new Error('Profile state domain version is invalid')
|
||||
}
|
||||
}
|
||||
|
||||
export function profileStateDomainsFromSerializedState(
|
||||
payload: Buffer
|
||||
): ProfileStateDomainMutation[] {
|
||||
const serialized = payload.toString('utf8')
|
||||
if (!Buffer.from(serialized, 'utf8').equals(payload)) {
|
||||
throw new Error('Profile state payload is not valid UTF-8')
|
||||
}
|
||||
return Object.entries(parseProfileStateRoot(serialized)).map(([domain, value]) => {
|
||||
const payload = JSON.stringify(value)
|
||||
if (payload === undefined) {
|
||||
throw new Error(`Profile state domain payload is not serializable: ${domain}`)
|
||||
}
|
||||
return { domain, payload }
|
||||
})
|
||||
}
|
||||
|
||||
@@ -5,7 +5,7 @@ import {
|
||||
PROFILE_STATE_DOCUMENT_VERSION,
|
||||
PROFILE_STATE_META_REVISION
|
||||
} from '../database/profile-state-database-schema'
|
||||
import './profile-state-documents'
|
||||
import { replaceProfileStateDomainsInTransaction } from './profile-state-documents'
|
||||
import {
|
||||
ProfileStateRevisionConflictError,
|
||||
isRecord,
|
||||
@@ -32,6 +32,9 @@ import {
|
||||
validateProfileStateDomainTransaction,
|
||||
type PreparedProfileStateMutation
|
||||
} from './profile-state-domain-write-validation'
|
||||
import type { ProfileStateDomainPatch } from './profile-state-domain-patches'
|
||||
import { preparePatchedProfileStateDomains } from './profile-state-patched-domain-preparation'
|
||||
import { buildCompleteDocumentReplacements } from './profile-state-complete-replacements'
|
||||
|
||||
export type ProfileStateDomainReplacement = {
|
||||
domain: string
|
||||
@@ -51,8 +54,12 @@ export type ProfileStateDomainTransaction = {
|
||||
replacements: readonly ProfileStateDomainMutation[]
|
||||
/** Enable selective run-row updates from Store's changed projection. */
|
||||
automationRunsAfter?: readonly unknown[]
|
||||
patches?: readonly ProfileStateDomainPatch[]
|
||||
complete?: boolean
|
||||
}
|
||||
|
||||
export type ProfileStateDomainChanges = Omit<ProfileStateDomainTransaction, 'expectedRevision'>
|
||||
|
||||
export type ProfileStateDomainWriteResult = {
|
||||
changed: boolean
|
||||
revision: number
|
||||
@@ -86,15 +93,51 @@ export function writeProfileStateDomains(
|
||||
db: Database.Database,
|
||||
transaction: ProfileStateDomainTransaction
|
||||
): ProfileStateDomainTransactionResult {
|
||||
validateProfileStateDomainTransaction(transaction)
|
||||
const patches = validateProfileStateDomainTransaction(transaction)
|
||||
|
||||
const prepared = transaction.replacements.map(prepareProfileStateDomainMutation)
|
||||
const prepared = transaction.replacements.map((replacement) =>
|
||||
prepareProfileStateDomainMutation(replacement)
|
||||
)
|
||||
|
||||
return withProfileStateWriteTransaction(db, () => {
|
||||
const actualRevision = readProfileStateRevision(db)
|
||||
if (actualRevision !== transaction.expectedRevision) {
|
||||
throw new ProfileStateRevisionConflictError(transaction.expectedRevision, actualRevision)
|
||||
}
|
||||
const mutations = [
|
||||
...prepared,
|
||||
...preparePatchedProfileStateDomains(db, patches, actualRevision)
|
||||
]
|
||||
if (transaction.complete) {
|
||||
const present = [
|
||||
...patches.map(({ domain }) => domain),
|
||||
...(transaction.automationRunsAfter === undefined ? [] : ['automationRuns'])
|
||||
]
|
||||
const complete = buildCompleteDocumentReplacements(db, mutations, present)
|
||||
if (actualRevision === 0 || (complete.length === 0 && present.length === 0)) {
|
||||
const entries = mutations.flatMap(({ domain, payload }) =>
|
||||
payload === null ? [] : [{ domain, payload }]
|
||||
)
|
||||
if (transaction.automationRunsAfter !== undefined) {
|
||||
entries.push({
|
||||
domain: 'automationRuns',
|
||||
payload: JSON.stringify(transaction.automationRunsAfter)
|
||||
})
|
||||
}
|
||||
const revision = replaceProfileStateDomainsInTransaction(
|
||||
db,
|
||||
entries,
|
||||
actualRevision,
|
||||
Date.now()
|
||||
)
|
||||
return { changed: true, revision, changedDomains: entries.map(({ domain }) => domain) }
|
||||
}
|
||||
mutations.push(
|
||||
...complete
|
||||
.slice(mutations.length)
|
||||
.map((mutation) => prepareProfileStateDomainMutation(mutation))
|
||||
)
|
||||
}
|
||||
const currentAutomationRuns = readCurrentAutomationRunsState(db, actualRevision)
|
||||
|
||||
const updates: PreparedProfileStateMutation[] = []
|
||||
@@ -154,7 +197,7 @@ export function writeProfileStateDomains(
|
||||
updates.push(mutation)
|
||||
}
|
||||
}
|
||||
for (const mutation of prepared) {
|
||||
for (const mutation of mutations) {
|
||||
admit(mutation)
|
||||
}
|
||||
if (transaction.automationRunsAfter !== undefined) {
|
||||
@@ -218,7 +261,8 @@ export function writeProfileStateDomains(
|
||||
revision: nextRevision,
|
||||
updatedAt,
|
||||
legacyRevision: splitState.legacyRevision,
|
||||
diffExisting: true
|
||||
diffExisting: true,
|
||||
existingParts: mutation.existingParts
|
||||
})
|
||||
) {
|
||||
continue
|
||||
|
||||
+69
@@ -0,0 +1,69 @@
|
||||
import type Database from '../../../sqlite/sync-database'
|
||||
import { assertProfileStateDocumentRevision } from '../database/profile-state-revision'
|
||||
import {
|
||||
readSplitDomainWriteState,
|
||||
readProfileStateSplitDocument,
|
||||
type ProfileStateExistingDocumentParts
|
||||
} from '../document-parts/profile-state-document-parts-storage'
|
||||
import { validateProfileStateDocumentRow } from './profile-state-document-validation'
|
||||
import {
|
||||
applyProfileStateDomainPatch,
|
||||
type PreparedProfileStateDomainPatch
|
||||
} from './profile-state-domain-patches'
|
||||
import {
|
||||
prepareProfileStateDomainMutation,
|
||||
type PreparedProfileStateMutation
|
||||
} from './profile-state-domain-write-validation'
|
||||
|
||||
export function preparePatchedProfileStateDomains(
|
||||
db: Database.Database,
|
||||
patches: readonly PreparedProfileStateDomainPatch[],
|
||||
actualRevision: number
|
||||
): PreparedProfileStateMutation[] {
|
||||
const mutations: PreparedProfileStateMutation[] = []
|
||||
for (const patch of patches) {
|
||||
const splitState = readSplitDomainWriteState(db, patch.domain)
|
||||
if (splitState.kind === 'parts') {
|
||||
assertProfileStateDocumentRevision(splitState.revision, actualRevision, patch.domain)
|
||||
if (patch.operations.length === 0) {
|
||||
continue
|
||||
}
|
||||
}
|
||||
let base: unknown
|
||||
let existingParts: ProfileStateExistingDocumentParts | undefined
|
||||
if (splitState.kind === 'parts') {
|
||||
existingParts = new Map()
|
||||
const document = readProfileStateSplitDocument(
|
||||
db,
|
||||
patch.domain,
|
||||
splitState.legacyRevision,
|
||||
existingParts
|
||||
)
|
||||
if (!document) {
|
||||
throw new Error('Profile state patch lost its authoritative parts')
|
||||
}
|
||||
base = JSON.parse(document.payload)
|
||||
} else {
|
||||
const row = db
|
||||
.prepare(
|
||||
`SELECT domain, payload, domain_version, revision, updated_at, content_hash
|
||||
FROM profile_state_documents WHERE domain = ?`
|
||||
)
|
||||
.get(patch.domain)
|
||||
if (row !== undefined) {
|
||||
const document = validateProfileStateDocumentRow(row, { retainParsedValue: true })
|
||||
assertProfileStateDocumentRevision(document.revision, actualRevision, patch.domain)
|
||||
base = document.value
|
||||
}
|
||||
}
|
||||
const value = applyProfileStateDomainPatch(base, patch)
|
||||
const payload = JSON.stringify(value)
|
||||
if (payload !== undefined) {
|
||||
mutations.push({
|
||||
...prepareProfileStateDomainMutation({ domain: patch.domain, payload }, { value }),
|
||||
existingParts
|
||||
})
|
||||
}
|
||||
}
|
||||
return mutations
|
||||
}
|
||||
@@ -3,6 +3,7 @@ import type {
|
||||
ProfileStateAuthority,
|
||||
ProfileStateAuthorityInitialState,
|
||||
ProfileStateDomainReplacement,
|
||||
PreparedProfileStateWrite,
|
||||
ProfileStateMaintenance
|
||||
} from '../loading-store/profile-writing/profile-state-authority'
|
||||
import {
|
||||
@@ -16,15 +17,16 @@ import {
|
||||
openWritableProfileStateDatabase,
|
||||
openProfileStateDatabaseReadOnly
|
||||
} from './database/profile-state-database'
|
||||
import { writeProfileStateDomains } from './documents/profile-state-domain-writes'
|
||||
import {
|
||||
writeProfileStateDomains,
|
||||
type ProfileStateDomainChanges
|
||||
} from './documents/profile-state-domain-writes'
|
||||
import {
|
||||
prepareProfileStateDomainMutation,
|
||||
profileStateDomainsFromSerializedState,
|
||||
validateProfileStateDomainTransaction
|
||||
} from './documents/profile-state-domain-write-validation'
|
||||
import {
|
||||
parseProfileStateRoot,
|
||||
ProfileStateRevisionConflictError
|
||||
} from './documents/profile-state-document-validation'
|
||||
import { ProfileStateRevisionConflictError } from './documents/profile-state-document-validation'
|
||||
import { assertProfileStateRevisionOnDisk } from './database/profile-state-revision-readmission'
|
||||
import {
|
||||
quarantineProfileStateDatabase,
|
||||
@@ -145,6 +147,19 @@ export class ProfileStateSqliteAuthority implements ProfileStateAuthority {
|
||||
replacements: readonly ProfileStateDomainReplacement[],
|
||||
automationRunsAfter?: readonly unknown[]
|
||||
): void {
|
||||
this.writeDomainTransaction({ replacements, automationRunsAfter })
|
||||
}
|
||||
|
||||
writePreparedState(write: PreparedProfileStateWrite): void {
|
||||
this.writeDomainTransaction({
|
||||
replacements: write.replacements,
|
||||
patches: write.patches,
|
||||
complete: write.complete,
|
||||
automationRunsAfter: write.runPayloads?.map((payload): unknown => JSON.parse(payload))
|
||||
})
|
||||
}
|
||||
|
||||
private writeDomainTransaction(transaction: ProfileStateDomainChanges): void {
|
||||
this.assertActive()
|
||||
if (this.observedRevision === undefined) {
|
||||
// Direct callers must establish a revision fence even without Store's initial read.
|
||||
@@ -153,9 +168,8 @@ export class ProfileStateSqliteAuthority implements ProfileStateAuthority {
|
||||
const opened = this.openWritableDatabase()
|
||||
this.legacyDocuments.catchUp(this.writableDatabase?.db)
|
||||
this.observedRevision = writeProfileStateDomains(opened.db, {
|
||||
expectedRevision: this.observedRevision ?? 0,
|
||||
replacements,
|
||||
automationRunsAfter
|
||||
...transaction,
|
||||
expectedRevision: this.observedRevision ?? 0
|
||||
}).revision
|
||||
this.legacyDocuments.afterWrite(() => this.writableDatabase?.db)
|
||||
}
|
||||
@@ -176,20 +190,7 @@ export class ProfileStateSqliteAuthority implements ProfileStateAuthority {
|
||||
|
||||
writeSerializedState(payload: Buffer): void {
|
||||
this.assertActive()
|
||||
const serialized = payload.toString('utf8')
|
||||
if (!Buffer.from(serialized, 'utf8').equals(payload)) {
|
||||
throw new Error('Profile state payload is not valid UTF-8')
|
||||
}
|
||||
const parsed = parseProfileStateRoot(serialized)
|
||||
this.writeCompleteSerializedDomains(
|
||||
Object.entries(parsed).map(([domain, value]) => {
|
||||
const payload = JSON.stringify(value)
|
||||
if (payload === undefined) {
|
||||
throw new Error(`Profile state domain payload is not serializable: ${domain}`)
|
||||
}
|
||||
return { domain, payload }
|
||||
})
|
||||
)
|
||||
this.writeCompleteSerializedDomains(profileStateDomainsFromSerializedState(payload))
|
||||
}
|
||||
|
||||
writeCompleteSerializedDomains(replacements: readonly ProfileStateDomainReplacement[]): void {
|
||||
|
||||
@@ -2,13 +2,17 @@ import type { AutomationRun } from '../../../shared/automations/history/automati
|
||||
import type {
|
||||
AsyncProfileStateAuthority,
|
||||
ProfileStateDomainReplacement,
|
||||
PreparedProfileStateWrite,
|
||||
ProfileStateMaintenance
|
||||
} from '../loading-store/profile-writing/profile-state-authority'
|
||||
import { ProfileStateBackupRotation } from './backup/profile-state-backup-rotation'
|
||||
import { runProfileStateBackupWorker } from './backup/profile-state-backup-worker'
|
||||
import { quarantineProfileStateDatabase } from './recovery/profile-state-database-quarantine'
|
||||
import { ProfileStateWriteWorkerClient } from './writer/profile-state-writer-worker-client'
|
||||
import { ProfileStateWriterError } from './writer/profile-state-writer-errors'
|
||||
import {
|
||||
ProfileStateWriterError,
|
||||
closedProfileStateWriterError
|
||||
} from './writer/profile-state-writer-errors'
|
||||
import type { ProfileStateWriterInitialization } from './writer/profile-state-writer-protocol'
|
||||
import {
|
||||
ProfileStateWriterReopenPolicy,
|
||||
@@ -62,7 +66,7 @@ export class ProfileStateWorkerAuthority implements AsyncProfileStateAuthority {
|
||||
} else if (this.stoppedFailure) {
|
||||
throw this.stoppedFailure
|
||||
} else if (this.closing) {
|
||||
throw closedWriterError()
|
||||
throw closedProfileStateWriterError()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -89,6 +93,10 @@ export class ProfileStateWorkerAuthority implements AsyncProfileStateAuthority {
|
||||
return this.run((writer) => writer.writeSerializedDomains(replacements)).then(() => {})
|
||||
}
|
||||
|
||||
writePreparedState(write: PreparedProfileStateWrite): Promise<void> {
|
||||
return this.run((writer) => writer.writePreparedState(write)).then(() => {})
|
||||
}
|
||||
|
||||
writeSerializedAutomationRuns(
|
||||
replacements: readonly ProfileStateDomainReplacement[],
|
||||
runs: readonly AutomationRun[]
|
||||
@@ -226,7 +234,7 @@ export class ProfileStateWorkerAuthority implements AsyncProfileStateAuthority {
|
||||
throw this.stoppedFailure
|
||||
}
|
||||
if (this.closing) {
|
||||
throw closedWriterError()
|
||||
throw closedProfileStateWriterError()
|
||||
}
|
||||
this.reopenPolicy.admit()
|
||||
const writer = new ProfileStateWriteWorkerClient(
|
||||
@@ -333,11 +341,3 @@ export class ProfileStateWorkerAuthority implements AsyncProfileStateAuthority {
|
||||
return this.stoppedRevision
|
||||
}
|
||||
}
|
||||
|
||||
function closedWriterError(): ProfileStateWriterError {
|
||||
return new ProfileStateWriterError(
|
||||
'profile-state-writer-closed',
|
||||
'Profile state writer is closing',
|
||||
'known-failure'
|
||||
)
|
||||
}
|
||||
|
||||
@@ -78,6 +78,23 @@ function readState(input: ReturnType<typeof options>) {
|
||||
}
|
||||
|
||||
describe('live profile authority admission', () => {
|
||||
it('saves split-domain changes through the real worker and reads them after graceful restart', async () => {
|
||||
const input = options()
|
||||
const first = await open(input)
|
||||
first.store.setWorkspaceSession({ ...first.store.getWorkspaceSession(), activeTabId: 'first' })
|
||||
await first.store.flushPendingOrThrowAsync()
|
||||
first.store.getWorkspaceSession().activeTabId = 'getter-only-final'
|
||||
await first.store.freezeWritesAsync()
|
||||
const reopened = await open(input)
|
||||
expect(reopened.store.getWorkspaceSession().activeTabId).toBe('getter-only-final')
|
||||
reopened.store.setWorkspaceSession({
|
||||
...reopened.store.getWorkspaceSession(),
|
||||
activeTabId: 'after-restart'
|
||||
})
|
||||
await reopened.store.freezeWritesAsync()
|
||||
expect(readState(input).workspaceSession.activeTabId).toBe('after-restart')
|
||||
})
|
||||
|
||||
it('never adopts a competing revision between bootstrap and worker readiness', async () => {
|
||||
const input = options()
|
||||
const original = ProfileStateSqliteAuthority.prototype.retireForWorker
|
||||
|
||||
@@ -116,3 +116,11 @@ function encodeOpenErrorCause(cause: unknown): ProfileStateWriterErrorData['caus
|
||||
...(errcode !== undefined ? { errcode } : {})
|
||||
}
|
||||
}
|
||||
|
||||
export function closedProfileStateWriterError(): ProfileStateWriterError {
|
||||
return new ProfileStateWriterError(
|
||||
'profile-state-writer-closed',
|
||||
'Profile state writer is closing',
|
||||
'known-failure'
|
||||
)
|
||||
}
|
||||
|
||||
@@ -1,4 +1,7 @@
|
||||
import type { ProfileStateDomainReplacement } from '../../loading-store/profile-writing/profile-state-authority'
|
||||
import type {
|
||||
ProfileStateDomainReplacement,
|
||||
PreparedProfileStateWrite
|
||||
} from '../../loading-store/profile-writing/profile-state-authority'
|
||||
import { isRecord } from '../documents/profile-state-document-validation'
|
||||
|
||||
export type ProfileStateWriterInitialization = {
|
||||
@@ -9,6 +12,7 @@ export type ProfileStateWriterInitialization = {
|
||||
|
||||
export type ProfileStateWriterCommand =
|
||||
| { command: 'write-state'; payload: Uint8Array }
|
||||
| { command: 'write-prepared'; write: PreparedProfileStateWrite }
|
||||
| {
|
||||
command: 'write-complete' | 'write-domains'
|
||||
replacements: readonly ProfileStateDomainReplacement[]
|
||||
@@ -65,11 +69,45 @@ function isReplacement(value: unknown): value is ProfileStateDomainReplacement {
|
||||
)
|
||||
}
|
||||
|
||||
function isStringArray(value: unknown): value is string[] {
|
||||
return Array.isArray(value) && value.every((entry: unknown) => typeof entry === 'string')
|
||||
}
|
||||
|
||||
function isPatchOperation(value: unknown): boolean {
|
||||
return (
|
||||
isRecord(value) &&
|
||||
isStringArray(value.path) &&
|
||||
(value.kind === 'remove' ||
|
||||
(value.kind === 'set' && typeof value.payload === 'string') ||
|
||||
(value.kind === 'order' && isStringArray(value.keys)))
|
||||
)
|
||||
}
|
||||
|
||||
function isPreparedWrite(value: unknown): value is PreparedProfileStateWrite {
|
||||
return (
|
||||
isRecord(value) &&
|
||||
Array.isArray(value.replacements) &&
|
||||
value.replacements.every(isReplacement) &&
|
||||
Array.isArray(value.patches) &&
|
||||
value.patches.every(
|
||||
(patch: unknown) =>
|
||||
isRecord(patch) &&
|
||||
typeof patch.domain === 'string' &&
|
||||
Array.isArray(patch.operations) &&
|
||||
patch.operations.every(isPatchOperation)
|
||||
) &&
|
||||
(value.complete === undefined || typeof value.complete === 'boolean') &&
|
||||
(value.runPayloads === undefined || isStringArray(value.runPayloads))
|
||||
)
|
||||
}
|
||||
|
||||
export function isProfileStateWriterRequest(value: unknown): value is ProfileStateWriterRequest {
|
||||
if (!isRecord(value) || !isProfileStateRevision(value.id) || value.id === 0) {
|
||||
return false
|
||||
}
|
||||
switch (value.command) {
|
||||
case 'write-prepared':
|
||||
return isPreparedWrite(value.write)
|
||||
case 'write-state':
|
||||
return value.payload instanceof Uint8Array
|
||||
case 'assert-revision':
|
||||
|
||||
@@ -1,9 +1,16 @@
|
||||
import type { ProfileStateDomainReplacement } from '../../loading-store/profile-writing/profile-state-authority'
|
||||
import type {
|
||||
ProfileStateDomainReplacement,
|
||||
PreparedProfileStateWrite
|
||||
} from '../../loading-store/profile-writing/profile-state-authority'
|
||||
import { ProfileStateWriterConnection } from './profile-state-writer-connection'
|
||||
import './profile-state-writer-worker-path'
|
||||
import './profile-state-writer-errors'
|
||||
|
||||
export class ProfileStateWriteWorkerClient extends ProfileStateWriterConnection {
|
||||
writePreparedState(write: PreparedProfileStateWrite): Promise<number> {
|
||||
return this.dispatch({ command: 'write-prepared', write }).then((response) => response.revision)
|
||||
}
|
||||
|
||||
assertWritable(): void {
|
||||
this.assertDispatchable()
|
||||
}
|
||||
|
||||
@@ -40,6 +40,9 @@ async function execute(request: ProfileStateWriterRequest): Promise<ProfileState
|
||||
}
|
||||
let exportedRevision: number | null | undefined
|
||||
switch (request.command) {
|
||||
case 'write-prepared':
|
||||
authority.writePreparedState(request.write)
|
||||
break
|
||||
case 'write-state':
|
||||
authority.writeSerializedState(Buffer.from(request.payload))
|
||||
break
|
||||
|
||||
Reference in New Issue
Block a user