Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(cordis): make config reload transactional

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-30 03:11:16 +08:00
parent f8eb6a9b84
commit 195f7fa9af
33 changed files with 1020 additions and 268 deletions
Download Patch File Download Diff File Expand all files Collapse all files

192
vendor/loader/src/config/entry.ts vendored
View File

@@ -21,6 +21,11 @@ export interface EntryOptions {
inject?: Inject | null
}
function updateError(stage: 'import' | 'dispose' | 'apply' | 'rollback', options: EntryOptions, cause: unknown) {
const detail = cause instanceof Error ? cause.message : String(cause)
return new Error(`failed to ${stage} loader entry ${options.id} (${options.name}): ${detail}`, { cause })
}
function takeEntries(object: {}, keys: string[]) {
const result: [string, any][] = []
for (const key of keys) {
@@ -38,6 +43,11 @@ function sortKeys<T extends {}>(object: T, prepend = ['id', 'name'], append = ['
return Object.assign(object, Object.fromEntries([...part1, ...rest, ...part2]))
}
function replaceKeys<T extends {}>(target: T, source: T): T {
for (const key of Object.keys(target)) Reflect.deleteProperty(target, key)
return Object.assign(target, source)
}
/** One configured plugin node inside an `EntryTree`. */
export class Entry {
static readonly key = Symbol.for('cordis.entry')
@@ -51,6 +61,7 @@ export class Entry {
public subtree?: EntryTree
_initTask?: Promise<void>
_disposing = 0
constructor(public loader: Loader) {
this.ctx = loader.ctx.extend({ [Entry.key]: this })
@@ -71,13 +82,18 @@ export class Entry {
/** True when this entry or any owning parent entry is disabled. */
get disabled() {
return this._disabled(this.options)
}
private _disabled(options: EntryOptions) {
// group is always enabled
if (this.options.group) return false
let entry: Entry | undefined = this
do {
if (options.group) return false
if (options.disabled) return true
let entry = this.parent.ctx.fiber.entry
while (entry) {
if (entry.options.disabled) return true
entry = entry.parent.ctx.fiber.entry
} while (entry)
}
return false
}
@@ -90,12 +106,12 @@ export class Entry {
return interpolate(this.ctx, this.options.config)
}
private _patchContext(diff: string[]) {
this.context.waterfall('loader/patch-context', this, () => {
private async _patchContext(diff: string[]) {
await this.context.waterfall('loader/patch-context', this, async () => {
Object.setPrototypeOf(this.ctx, this.parent.ctx)
if (this.fiber?.uid && (diff.includes('config') || this.options.group)) {
this.fiber.update(this._resolveConfig(this.fiber.runtime!.callback), true)
await this.fiber.update(this._resolveConfig(this.fiber.runtime!.callback), true)
}
})
}
@@ -106,41 +122,122 @@ export class Entry {
await this.init()
}
async _dispose(fiber = this.fiber) {
if (!fiber) return
if (this.fiber === fiber) this.fiber = undefined
this._disposing += 1
try {
await fiber.dispose()
} finally {
this._disposing -= 1
}
}
/** Merge new options, restart as needed, and persist through the parent tree. */
async update(options: Partial<EntryOptions>, create = false, force = false) {
const legacy = { ...this.options }
// step 1: update options
if (create) {
this.options = options as EntryOptions
} else {
const previousOptions = this.options
const legacy = { ...previousOptions }
const candidate = create ? options as EntryOptions : { ...previousOptions }
if (!create) {
for (const [key, value] of Object.entries(options)) {
if (isNullable(value)) {
delete this.options[key]
delete candidate[key as keyof EntryOptions]
} else {
this.options[key] = value
candidate[key as keyof EntryOptions] = value as never
}
}
}
sortKeys(this.options)
sortKeys(candidate)
// step 2: execute
if (this.disabled) {
this.fiber?.dispose()
const diff = Object
.keys({ ...candidate, ...legacy })
.filter(key => !deepEqual(candidate[key as keyof EntryOptions], legacy[key as keyof EntryOptions]))
if (!diff.length && !force) return
const commit = () => {
if (create) return
this.options = replaceKeys(previousOptions, candidate)
}
const previous = this.fiber
if (!previous?.uid) {
this.fiber = undefined
this.options = candidate
try {
if (!this._disabled(candidate)) await this.init()
} catch (error) {
this.options = previousOptions
throw error
}
commit()
return
}
// step 3: check if options are changed
if (this.fiber?.uid) {
const diff = Object
.keys({ ...this.options, ...legacy })
.filter(key => !deepEqual(this.options[key], legacy[key]))
if (!diff.length && !force) return
if (this._disabled(candidate)) {
this.options = candidate
try {
await this._dispose(previous)
} catch (error) {
this.options = previousOptions
throw updateError('dispose', candidate, error)
}
commit()
this.context.emit('loader/partial-dispose', this, legacy, true)
this._patchContext(diff)
} else {
await this.init()
return
}
const replace = diff.some(key => key === 'name' || key === 'inject' || key === 'group')
if (!replace) {
this.options = candidate
try {
await this._patchContext(diff)
} catch (error) {
this.options = previousOptions
try {
await this._patchContext(diff)
} catch (rollbackError) {
throw updateError('rollback', legacy, new AggregateError([error, rollbackError]))
}
this.context.emit('loader/partial-dispose', this, candidate, true)
throw updateError('apply', candidate, error)
}
commit()
this.context.emit('loader/partial-dispose', this, legacy, true)
return
}
let plugin: any
try {
plugin = diff.includes('name')
? this.loader.unwrapExports(await this.parent.tree.import(candidate.name, this.getOuterStack))
: previous.runtime!.callback
} catch (error) {
throw updateError('import', candidate, error)
}
const previousPlugin = previous.runtime!.callback
this.options = candidate
try {
await this._dispose(previous)
} catch (error) {
this.options = previousOptions
throw updateError('dispose', candidate, error)
}
try {
await this._start(plugin)
} catch (error) {
this.options = previousOptions
try {
await this._start(previousPlugin)
} catch (rollbackError) {
throw updateError('rollback', legacy, new AggregateError([error, rollbackError]))
}
this.context.emit('loader/partial-dispose', this, candidate, true)
throw updateError('apply', candidate, error)
}
commit()
this.context.emit('loader/partial-dispose', this, legacy, true)
}
getOuterStack = () => {
@@ -159,26 +256,39 @@ export class Entry {
await (this._initTask ??= this._init())
} finally {
this._initTask = undefined
if (!this.loader.getTasks().length) this.ctx.reflect.notify(['loader'])
}
this.fiber?.await().finally(() => {
if (this.loader.getTasks().length) return
this.ctx.reflect.notify(['loader'])
})
await this.fiber?.await()
}
private async _init() {
let exports: any
let plugin: any
try {
exports = await this.parent.tree.import(this.options.name, this.getOuterStack)
plugin = this.loader.unwrapExports(await this.parent.tree.import(this.options.name, this.getOuterStack))
} catch (error) {
this.ctx.logger.error(error)
return
} finally {
this._initTask = undefined
throw updateError('import', this.options, error)
}
const plugin = this.loader.unwrapExports(exports)
this._patchContext([])
try {
await this._start(plugin)
} catch (error) {
throw updateError('apply', this.options, error)
}
}
private async _start(plugin: any) {
let fiber: Fiber | undefined
try {
fiber = await this._create(plugin)
await fiber.await()
} catch (error) {
await this._dispose(fiber)
throw error
}
}
private async _create(plugin: any): Promise<Fiber> {
await this._patchContext([])
this.loader.showLog(this, 'apply')
this.fiber = this.ctx.registry.plugin(plugin, this._resolveConfig(plugin), this.getOuterStack)
return this.fiber = this.ctx.registry.plugin(plugin, this._resolveConfig(plugin), this.getOuterStack)
}
}

72
vendor/loader/src/config/group.ts vendored
View File

@@ -19,12 +19,23 @@ export class EntryGroup {
async create(options: Omit<EntryOptions, 'id'>) {
const id = this.tree.ensureId(options)
const entry: Entry = this.tree.store[id] ??= new Entry(this.ctx.loader)
const existing = this.tree.store[id]
const entry: Entry = existing ?? (this.tree.store[id] = new Entry(this.ctx.loader))
const previousParent = entry.parent
// Entry may be moved from another group,
// so we need to update the parent reference.
entry.parent = this
// Use `create: true` to replace existing entry.options.
await entry.update(options, true, true)
try {
await entry.update(options, true, true)
} catch (error) {
if (existing) {
entry.parent = previousParent
} else {
delete this.tree.store[id]
}
throw error
}
return entry.id
}
@@ -34,10 +45,10 @@ export class EntryGroup {
if (index >= 0) config.splice(index, 1)
}
remove(id: string, isDispose = false) {
async remove(id: string, isDispose = false) {
const entry = this.tree.store[id]
if (!entry) return
entry.fiber?.dispose()
await entry._dispose()
if (!isDispose) {
this.unlink(entry.options)
}
@@ -47,26 +58,47 @@ export class EntryGroup {
async update(config: EntryOptions[]) {
const oldConfig = this.data as EntryOptions[]
this.data = config
const seen = new Set<string>()
for (const options of config) {
const id = this.tree.ensureId(options)
if (seen.has(id)) throw new TypeError(`duplicate loader entry id: ${id}`)
seen.add(id)
}
const oldMap = Object.fromEntries(oldConfig.map(options => [options.id, options]))
const newMap = Object.fromEntries(config.map(options => [options.id ?? Symbol('anonymous'), options]))
const newMap = Object.fromEntries(config.map(options => [options.id, options]))
// update inner plugins
const ids = Reflect.ownKeys({ ...oldMap, ...newMap }) as string[]
await Promise.all(ids.map(async (id) => {
if (newMap[id]) {
await this.create(newMap[id]).catch((error) => {
this.ctx.logger.error(error)
})
} else {
this.remove(id)
try {
for (const options of config) await this.create(options)
for (const id of Object.keys(oldMap)) {
if (!newMap[id]) await this.remove(id, true)
}
}))
this.data = config
} catch (error) {
const rollbackErrors: unknown[] = []
for (const id of Object.keys(newMap).reverse()) {
if (oldMap[id]) continue
try {
await this.remove(id, true)
} catch (rollbackError) {
rollbackErrors.push(rollbackError)
}
}
for (const options of oldConfig) {
try {
await this.create(options)
} catch (rollbackError) {
rollbackErrors.push(rollbackError)
}
}
this.data = oldConfig
if (rollbackErrors.length) throw new AggregateError([error, ...rollbackErrors], 'loader entry rollback failed')
throw error
}
}
stop() {
async stop() {
for (const options of this.data) {
this.remove(options.id, true)
await this.remove(options.id, true)
}
}
}
@@ -78,9 +110,7 @@ export class Group extends EntryGroup {
constructor(public ctx: Context, public config: EntryOptions[]) {
super(ctx, ctx.fiber.entry!.parent.tree)
ctx.on('internal/update', (config) => {
this.update(config)
})
ctx.on('internal/update', config => this.update(config))
}
async* [Service.init]() {

4
vendor/loader/src/config/isolate.ts vendored
View File

@@ -93,7 +93,7 @@ export default function isolate(ctx: Context) {
entry.ctx[Context.isolate] = Object.create(entry.ctx[Context.isolate])
})
ctx.on('loader/patch-context', (entry, next) => {
ctx.on('loader/patch-context', async (entry, next) => {
// step 1: generate new isolate map
const newMap: Dict<symbol> = Object.create(entry.parent.ctx[Context.isolate])
for (const name of Object.keys(entry.options.isolate ?? {})) {
@@ -126,7 +126,7 @@ export default function isolate(ctx: Context) {
swap(entry.ctx[Context.intercept], entry.options.intercept)
// step 4: reload fiber
next()
await next()
// step 5: replace service impl
for (const [symbol1, symbol2, flag1, flag2] of Object.values(diff)) {

53
vendor/loader/src/config/tree.ts vendored
View File

@@ -39,12 +39,27 @@ export abstract class EntryTree {
.filter(isNonNullable)
}
/** Wait until this tree has no pending import or lifecycle tasks. */
/**
* Wait until this tree has no active import or lifecycle tasks.
* @throws a settled fiber failure, or an aggregate when several fibers failed.
*/
async await() {
while (true) {
const tasks = this.getTasks()
if (!tasks.length) return
await Promise.allSettled(tasks)
if (tasks.length) {
await Promise.allSettled(tasks)
continue
}
const outcomes = await Promise.allSettled(
[...this.entries()].map(entry => entry.fiber?.await()),
)
const failures = outcomes
.filter((outcome): outcome is PromiseRejectedResult => outcome.status === 'rejected')
.map(outcome => outcome.reason)
if (failures.length === 1) throw failures[0]
if (failures.length > 1) throw new AggregateError(failures, 'loader fibers failed')
this.ctx.reflect.notify(['loader'])
if (!this.getTasks().length) return
}
}
@@ -81,15 +96,17 @@ export abstract class EntryTree {
/** Create an entry in the root group or a nested group. */
async create(options: Omit<EntryOptions, 'id'>, parent: string | null = null, position = Infinity) {
const group = this.resolveGroup(parent)
group.data.splice(position, 0, options as EntryOptions)
const id = await group.create(options)
const entry = this.resolve(id)
group.data.splice(position, 0, entry.options)
group.tree.write()
return group.create(options)
return id
}
/** Stop and remove an entry from its parent group. */
remove(id: string) {
async remove(id: string) {
const entry = this.resolve(id)
entry.parent.remove(id)
await entry.parent.remove(id)
entry.parent.tree.write()
}
@@ -97,15 +114,31 @@ export abstract class EntryTree {
async update(id: string, options: Omit<EntryOptions, 'id' | 'name'>, parent?: string | null, position?: number) {
const entry = this.resolve(id)
const source = entry.parent
const sourceIndex = source.data.indexOf(entry.options)
let target = source
if (parent !== undefined) {
const target = this.resolveGroup(parent)
target = this.resolveGroup(parent)
source.unlink(entry.options)
target.data.splice(position ?? Infinity, 0, entry.options)
target.tree.write()
entry.parent = target
}
try {
await entry.update(options, false, true)
} catch (error) {
if (parent !== undefined) {
target.unlink(entry.options)
source.data.splice(sourceIndex < 0 ? source.data.length : sourceIndex, 0, entry.options)
entry.parent = source
try {
await entry.update({}, false, true)
} catch (rollbackError) {
throw new AggregateError([error, rollbackError], `failed to roll back loader entry move ${id}`)
}
}
throw error
}
source.tree.write()
return entry.update(options, false, true)
if (target !== source) target.tree.write()
}
/** Import a plugin module from a specifier or `cordis:` builtin. */

11
vendor/loader/src/index.ts vendored
View File

@@ -24,7 +24,7 @@ declare module 'cordis' {
'loader/config-update'(): void
'loader/entry-init'(entry: Entry): void
'loader/partial-dispose'(entry: Entry, legacy: Partial<EntryOptions>, active: boolean): void
'loader/patch-context'(entry: Entry, next: () => void): void
'loader/patch-context'(entry: Entry, next: () => void | Promise<void>): void | Promise<void>
}
interface Context {
@@ -87,12 +87,12 @@ export class Loader extends EntryTree {
ctx.reflect.provide('loader', this, this[Service.check])
ctx.on('internal/update', function (config, noSave, next) {
ctx.on('internal/update', async function (config, noSave, next) {
if (!this.entry || noSave || this.parent.fiber?.entry === this.entry) return next()
await next()
const unparse = this.runtime?.Config?.['simplify']
this.entry.options.config = unparse ? unparse(config) : config
this.entry.parent.tree.write()
return next()
}, { global: true, prepend: true })
ctx.on('internal/update', function (config, _, next) {
@@ -129,9 +129,12 @@ export class Loader extends EntryTree {
// case 5: the entry's tree is being disposed
if (!fiber.entry.parent.tree.ctx.fiber.uid) return
// case 6: Loader is replacing or removing this exact fiber
if (fiber.entry._disposing) return
this.showLog(fiber.entry, 'unload')
// case 6: fiber is disposed by loader behavior
// case 7: fiber is disposed by loader behavior
// such as inject checker, config file update, ancestor group disable
if (fiber.entry.disabled) return