mirror of
https://github.com/shadowsocks/ShadowsocksX-NG.git
synced 2026-10-29 05:33:45 +00:00
refactor(runtime): 集中收敛意图与恢复策略
This commit is contained in:
@@ -154,49 +154,10 @@ extension ProxyRuntimeController {
|
||||
await deploy(document)
|
||||
}
|
||||
|
||||
/// 把「运行定义档」意图推到运行时:派生文档 → 收敛脊柱 apply(幂等跳过、
|
||||
/// 计划执行、健康门)。
|
||||
/// 普通部署意图;配置派生、执行与失败呈现由运行状态收敛 module 拥有。
|
||||
@discardableResult
|
||||
func deploy(_ sourceDocument: SslocalRuntimeDocument) async -> ConvergenceResult {
|
||||
if listenSettingsUnreadable || settingsUnreadable {
|
||||
await refuseDeployForUnreadableListenSettings()
|
||||
return .failed
|
||||
}
|
||||
let contract: PreparedRuntimeContract
|
||||
do {
|
||||
contract = try PreparedRuntimeContract(await runtimeDocument(sourceDocument, for: proxyMode))
|
||||
} catch RulePreparationError.superseded {
|
||||
return .superseded
|
||||
} catch {
|
||||
// 规则快照缺失/损坏:不静默退化成全局(issue #63 AC4)。
|
||||
RuntimeLog.emit(.runtimePersistFailed(detail: String(describing: error)))
|
||||
state = .serviceFailed(.runtimeFile)
|
||||
return .failed
|
||||
}
|
||||
return await deployPrepared(contract)
|
||||
}
|
||||
|
||||
@discardableResult
|
||||
func deployPrepared(_ contract: PreparedRuntimeContract) async -> ConvergenceResult {
|
||||
let document = contract.document
|
||||
var ticket = convergenceTicket()
|
||||
return await apply(
|
||||
document, ticket: &ticket,
|
||||
requiresReceipt: document.aclRuntime != nil,
|
||||
convergeProxyOnSuccess: true,
|
||||
preserveProxyOnFailure: false,
|
||||
proxyTail: .none, rollback: nil)
|
||||
}
|
||||
|
||||
/// D8「任何路径不静默改端口」:监听设置不可读时以占位出厂端口部署等于
|
||||
/// 系统擅自改端口——停止运行时并点名呈现,等用户在设置区修复(#33 接线)。
|
||||
private func refuseDeployForUnreadableListenSettings() async {
|
||||
RuntimeLog.emit(.activationFailed(reason: "listen settings unreadable"))
|
||||
_ = await execute(.stop, document: nil)
|
||||
lastDocument = nil
|
||||
skippedServers = []
|
||||
state = .launchFailed(.unreadableSettings)
|
||||
await systemProxyObserver.holdSystemProxyIntent()
|
||||
await convergeRuntime(.deployment(sourceDocument))
|
||||
}
|
||||
|
||||
/// 活动目标失效(issue #60):清除并持久化 nil,点名原因独立呈现;agent
|
||||
|
||||
@@ -21,7 +21,7 @@ struct RuntimeExecutionOutcome {
|
||||
/// 与推进时机:flow 分量在 execute 后以实际占用值推进,mode 切换路径在
|
||||
/// 派生后推进 preparation 分量(派生自身会推进它),rules 部署在推进 mode
|
||||
/// 代际之后捕获。
|
||||
struct ConvergenceTicket: Equatable {
|
||||
private struct ConvergenceTicket: Equatable {
|
||||
/// 并发流代际;每次 execute 之后由调用方以实际占用值推进。
|
||||
var flow: Int
|
||||
/// 模式切换代际;捕获后不再变化。
|
||||
@@ -32,7 +32,7 @@ struct ConvergenceTicket: Equatable {
|
||||
|
||||
extension ProxyRuntimeController {
|
||||
/// 以当前代际捕获收敛票据。
|
||||
func convergenceTicket() -> ConvergenceTicket {
|
||||
private func convergenceTicket() -> ConvergenceTicket {
|
||||
ConvergenceTicket(
|
||||
flow: flowGeneration, mode: modeChangeGeneration,
|
||||
preparation: runtimePreparationGeneration)
|
||||
@@ -40,7 +40,7 @@ extension ProxyRuntimeController {
|
||||
|
||||
/// 票据是否仍代表当前收敛:并发流 / 模式切换 / 派生三个代际均未前进,
|
||||
/// 且会话内 agent 意图仍开启。
|
||||
func convergenceIsCurrent(_ ticket: ConvergenceTicket) -> Bool {
|
||||
private func convergenceIsCurrent(_ ticket: ConvergenceTicket) -> Bool {
|
||||
ticket.flow == flowGeneration && ticket.mode == modeChangeGeneration
|
||||
&& ticket.preparation == runtimePreparationGeneration
|
||||
&& settings.agentEnabled
|
||||
@@ -51,7 +51,7 @@ extension ProxyRuntimeController {
|
||||
/// 存活、控制器处于健康运行态——计划层动作序列为空即幂等跳过。判定在
|
||||
/// 置 `starting` 之前进行,跳过路径不闪状态、不重走健康门;磁盘漂移或
|
||||
/// wrapper 失踪时动作序列自然非空,走完整 deploy 自愈。
|
||||
func convergencePlanIsEmpty(
|
||||
private func convergencePlanIsEmpty(
|
||||
_ document: SslocalRuntimeDocument, preparedContract: PreparedRuntimeContract?
|
||||
) -> Bool {
|
||||
switch state {
|
||||
@@ -96,46 +96,162 @@ extension ProxyRuntimeController {
|
||||
}
|
||||
}
|
||||
|
||||
/// 收敛脊柱 apply 层的结果。
|
||||
/// 一个意图入口隐藏派生、执行政策、恢复和失败呈现;调用者不组合执行选项。
|
||||
enum RuntimeConvergenceIntent {
|
||||
case deployment(SslocalRuntimeDocument)
|
||||
case modeTransition(ModeTransitionSnapshot)
|
||||
case savedRules
|
||||
}
|
||||
|
||||
/// 用户改变模式前的事实,由意图命令捕获;恢复计划只在收敛实现中构造。
|
||||
struct ModeTransitionSnapshot {
|
||||
let settings: ProxySettings
|
||||
let mode: ProxyMode
|
||||
let document: SslocalRuntimeDocument?
|
||||
let state: AgentRunState
|
||||
}
|
||||
|
||||
enum ConvergenceResult {
|
||||
/// 部署已执行且启动健康通过。
|
||||
case applied
|
||||
/// 派生契约与运行中运行时逐字节相同:不写契约、不发信号、不闪
|
||||
/// starting、不重走健康门与系统代理收敛(幂等跳过)。
|
||||
case unchanged
|
||||
/// 执行或健康门失败;回滚由调用方按 RollbackPlan 处理。
|
||||
/// 派生或部署失败;该意图要求的恢复与失败呈现已经处理。
|
||||
case failed
|
||||
/// 被更新的收敛取代:不得再变更任何状态。
|
||||
/// 被新意图取代,不得恢复或发布旧结果。
|
||||
case superseded
|
||||
}
|
||||
|
||||
/// 成功收敛后的系统代理收尾方式。
|
||||
enum ProxyConvergenceTail {
|
||||
/// 强制重收敛(mode 切换:系统代理值随模式 ACL 变化)。
|
||||
case forceApply
|
||||
/// 常规收敛(rules 提交)。
|
||||
case converge
|
||||
/// 无需收尾(deploy 路径由健康门自身的 convergeProxyOnSuccess 收敛)。
|
||||
case none
|
||||
@discardableResult
|
||||
func convergeRuntime(_ intent: RuntimeConvergenceIntent) async -> ConvergenceResult {
|
||||
switch intent {
|
||||
case .deployment(let source): return await convergeDeployment(source)
|
||||
case .modeTransition(let snapshot): return await convergeModeTransition(snapshot)
|
||||
case .savedRules: return await convergeSavedRules()
|
||||
}
|
||||
}
|
||||
|
||||
/// 收敛脊柱 apply 层:把一个已派生的运行时定义推到运行时——幂等跳过、
|
||||
/// 计划执行、启动健康门、成功后按参数收敛系统代理。失败不在此回滚:
|
||||
/// 返回 `.failed` 由调用方按 RollbackPlan 整体恢复。
|
||||
///
|
||||
/// 票据 inout 传参:execute / 派生后的代际推进对调用方可见,调用方
|
||||
/// 在 apply 之后的时效检查使用的是推进过的票据。
|
||||
///
|
||||
/// `starting` 在 execute 之前置位:`.run` 计划的动作全部同步执行,二者
|
||||
/// 之间不存在可见的挂起点,与旧路径「deploy 后置位」不可区分。
|
||||
func apply(
|
||||
private func convergeDeployment(_ source: SslocalRuntimeDocument) async -> ConvergenceResult {
|
||||
if listenSettingsUnreadable || settingsUnreadable {
|
||||
RuntimeLog.emit(.activationFailed(reason: "listen settings unreadable"))
|
||||
_ = await execute(.stop, document: nil)
|
||||
lastDocument = nil
|
||||
skippedServers = []
|
||||
state = .launchFailed(.unreadableSettings)
|
||||
await systemProxyObserver.holdSystemProxyIntent()
|
||||
return .failed
|
||||
}
|
||||
let contract: PreparedRuntimeContract
|
||||
do {
|
||||
contract = try PreparedRuntimeContract(await runtimeDocument(source, for: proxyMode))
|
||||
} catch RulePreparationError.superseded {
|
||||
return .superseded
|
||||
} catch {
|
||||
RuntimeLog.emit(.runtimePersistFailed(detail: String(describing: error)))
|
||||
state = .serviceFailed(.runtimeFile)
|
||||
return .failed
|
||||
}
|
||||
var ticket = convergenceTicket()
|
||||
return await apply(contract.document, ticket: &ticket, policy: .deployment)
|
||||
}
|
||||
|
||||
private func convergeModeTransition(_ snapshot: ModeTransitionSnapshot) async -> ConvergenceResult
|
||||
{
|
||||
var ticket = convergenceTicket()
|
||||
guard settings.agentEnabled, state != .off,
|
||||
let current = lastDocument ?? runtimeFileStore.loadDocument()
|
||||
else {
|
||||
systemProxyObserver.modeTransitionAwaitingRuntime()
|
||||
return .unchanged
|
||||
}
|
||||
let plan = RollbackPlan(
|
||||
document: snapshot.document ?? current, state: snapshot.state,
|
||||
payload: .modeTransition(
|
||||
mode: snapshot.mode, ruleDefaultAction: snapshot.settings.ruleDefaultAction),
|
||||
systemProxyIntentAtCapture: snapshot.settings.systemProxyEnabled,
|
||||
convergeProxyOnIntentChange: true)
|
||||
let next: SslocalRuntimeDocument
|
||||
do {
|
||||
next = try await runtimeDocument(current, for: proxyMode)
|
||||
} catch RulePreparationError.superseded {
|
||||
return .superseded
|
||||
} catch {
|
||||
RuntimeLog.emit(.runtimePersistFailed(detail: String(describing: error)))
|
||||
ticket.preparation = runtimePreparationGeneration
|
||||
await recoverModeTransition(plan, ticket: &ticket)
|
||||
return .failed
|
||||
}
|
||||
// 派生会推进 preparation;仅该意图在派生前捕获了收敛票据。
|
||||
ticket.preparation = runtimePreparationGeneration
|
||||
guard next.aclRuntime != current.aclRuntime else {
|
||||
// 保留原有特殊路径:无系统代理意图即结束;否则只检查健康,不部署或恢复。
|
||||
guard settings.systemProxyEnabled else { return .unchanged }
|
||||
state = .starting
|
||||
return await presentLaunchHealth(next) ? .applied : .failed
|
||||
}
|
||||
let result = await apply(next, ticket: &ticket, policy: .modeTransition)
|
||||
if case .failed = result { await recoverModeTransition(plan, ticket: &ticket) }
|
||||
return result
|
||||
}
|
||||
|
||||
private func recoverModeTransition(_ plan: RollbackPlan, ticket: inout ConvergenceTicket) async {
|
||||
let report = await restore(plan, ticket: &ticket)
|
||||
if report.runtimeHealthy == true, report.payloadFailureDescription != nil {
|
||||
state = .serviceFailed(.persistence)
|
||||
}
|
||||
}
|
||||
|
||||
private func convergeSavedRules() async -> ConvergenceResult {
|
||||
guard proxyMode == .rule, settings.agentEnabled, state != .off,
|
||||
let current = lastDocument ?? runtimeFileStore.loadDocument()
|
||||
else { return .unchanged }
|
||||
let revision = ruleDocuments.revision
|
||||
let plan = RollbackPlan(
|
||||
document: current, state: state, payload: .runtimeOnly,
|
||||
systemProxyIntentAtCapture: settings.systemProxyEnabled,
|
||||
convergeProxyOnIntentChange: false)
|
||||
let next: SslocalRuntimeDocument
|
||||
do {
|
||||
next = try await runtimeDocument(current, for: proxyMode)
|
||||
} catch RulePreparationError.superseded {
|
||||
return .superseded
|
||||
} catch {
|
||||
RuntimeLog.emit(.runtimePersistFailed(detail: String(describing: error)))
|
||||
ruleApplicationFailure = .service(.runtimeFile)
|
||||
return .failed
|
||||
}
|
||||
guard next.aclRuntime != current.aclRuntime || ruleApplicationFailure != nil else {
|
||||
ruleApplicationFailure = nil
|
||||
return .unchanged
|
||||
}
|
||||
modeChangeGeneration += 1
|
||||
var ticket = convergenceTicket()
|
||||
let result = await apply(next, ticket: &ticket, policy: .savedRules)
|
||||
switch result {
|
||||
case .applied, .unchanged:
|
||||
if revision == ruleDocuments.revision { ruleApplicationFailure = nil }
|
||||
case .superseded:
|
||||
break
|
||||
case .failed:
|
||||
let failure = ProxyRuntimeFacts(state: state).failure ?? .service(.runtimeFile)
|
||||
_ = await restore(plan, ticket: &ticket)
|
||||
if convergenceIsCurrent(ticket), revision == ruleDocuments.revision {
|
||||
ruleApplicationFailure = failure
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
/// 三种意图的执行差异只在 module 内选择,不能由调用者拼成任意组合。
|
||||
private enum DeploymentPolicy {
|
||||
case deployment
|
||||
case modeTransition
|
||||
case savedRules
|
||||
}
|
||||
|
||||
/// 共用执行脊柱;恢复在同一 module 的意图实现中,票据不越过意图 seam。
|
||||
private func apply(
|
||||
_ document: SslocalRuntimeDocument,
|
||||
ticket: inout ConvergenceTicket,
|
||||
requiresReceipt: Bool,
|
||||
convergeProxyOnSuccess: Bool,
|
||||
preserveProxyOnFailure: Bool,
|
||||
proxyTail: ProxyConvergenceTail,
|
||||
rollback: RollbackPlan?
|
||||
policy: DeploymentPolicy
|
||||
) async -> ConvergenceResult {
|
||||
let contract = try? PreparedRuntimeContract(document)
|
||||
if convergencePlanIsEmpty(document, preparedContract: contract) {
|
||||
@@ -152,17 +268,17 @@ extension ProxyRuntimeController {
|
||||
guard execution.succeeded else { return .failed }
|
||||
let healthy = await presentLaunchHealth(
|
||||
document,
|
||||
requiresReceipt: requiresReceipt,
|
||||
convergeProxyOnSuccess: convergeProxyOnSuccess,
|
||||
preserveProxyOnFailure: preserveProxyOnFailure, preparedContract: contract)
|
||||
requiresReceipt: policy != .deployment || document.aclRuntime != nil,
|
||||
convergeProxyOnSuccess: policy == .deployment,
|
||||
preserveProxyOnFailure: policy != .deployment, preparedContract: contract)
|
||||
guard convergenceIsCurrent(ticket) else { return .superseded }
|
||||
guard healthy else { return .failed }
|
||||
lastDocument = document
|
||||
ruleApplicationFailure = nil
|
||||
switch proxyTail {
|
||||
case .forceApply: await systemProxyObserver.convergeSystemProxy(forceApply: true)
|
||||
case .converge: await systemProxyObserver.convergeSystemProxy()
|
||||
case .none: break
|
||||
switch policy {
|
||||
case .modeTransition: await systemProxyObserver.convergeSystemProxy(forceApply: true)
|
||||
case .savedRules: await systemProxyObserver.convergeSystemProxy()
|
||||
case .deployment: break
|
||||
}
|
||||
return .applied
|
||||
}
|
||||
@@ -173,7 +289,8 @@ extension ProxyRuntimeController {
|
||||
/// 由计划层对注销/信号/注册的裁决保证与旧 restore 例程等价)→ 启动健康
|
||||
/// 门。载荷持久化失败不阻断文档恢复;文档恢复失败或健康不过即撤下系统
|
||||
/// 代理意图。
|
||||
func restore(_ plan: RollbackPlan, ticket: inout ConvergenceTicket) async -> RestoreReport {
|
||||
private func restore(_ plan: RollbackPlan, ticket: inout ConvergenceTicket) async -> RestoreReport
|
||||
{
|
||||
guard convergenceIsCurrent(ticket) else {
|
||||
return RestoreReport(payloadFailureDescription: nil, runtimeHealthy: nil)
|
||||
}
|
||||
@@ -235,7 +352,7 @@ extension ProxyRuntimeController {
|
||||
|
||||
/// 收敛失败时的整体回滚计划:恢复到哪个运行时文档、健康通过后呈现哪个
|
||||
/// 状态、以及除文档外还要重持久化的载荷。
|
||||
struct RollbackPlan {
|
||||
private struct RollbackPlan {
|
||||
/// 回滚载荷:除运行时文档外还要恢复的持久化事实。
|
||||
enum Payload {
|
||||
/// mode 切换:设置快照的模式与规则子选项回退(其余字段保留当前值),
|
||||
@@ -257,7 +374,7 @@ struct RollbackPlan {
|
||||
}
|
||||
|
||||
/// 回滚报告:载荷重持久化与旧运行时恢复的结果。
|
||||
struct RestoreReport {
|
||||
private struct RestoreReport {
|
||||
/// 载荷持久化失败的点名描述;nil = 成功。
|
||||
let payloadFailureDescription: String?
|
||||
/// 旧运行时是否恢复到健康;nil = 恢复未尝试(被弃权或被取代)。
|
||||
|
||||
@@ -126,57 +126,12 @@ extension ProxyRuntimeController {
|
||||
let generation = self.ruleApplicationGeneration
|
||||
do { try await self.ruleApplicationDelay() } catch { return }
|
||||
guard generation == self.ruleApplicationGeneration else { continue }
|
||||
await self.applySavedRuleDocument()
|
||||
await self.convergeRuntime(.savedRules)
|
||||
if generation == self.ruleApplicationGeneration { return }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func applySavedRuleDocument() async {
|
||||
guard proxyMode == .rule, settings.agentEnabled, state != .off,
|
||||
let currentDocument = lastDocument ?? runtimeFileStore.loadDocument()
|
||||
else { return }
|
||||
let revision = ruleDocuments.revision
|
||||
let plan = RollbackPlan(
|
||||
document: currentDocument, state: state,
|
||||
payload: .runtimeOnly,
|
||||
systemProxyIntentAtCapture: settings.systemProxyEnabled,
|
||||
convergeProxyOnIntentChange: false)
|
||||
let nextDocument: SslocalRuntimeDocument
|
||||
do {
|
||||
nextDocument = try await runtimeDocument(currentDocument, for: proxyMode)
|
||||
} catch RulePreparationError.superseded {
|
||||
return
|
||||
} catch {
|
||||
RuntimeLog.emit(.runtimePersistFailed(detail: String(describing: error)))
|
||||
ruleApplicationFailure = .service(.runtimeFile)
|
||||
return
|
||||
}
|
||||
guard nextDocument.aclRuntime != currentDocument.aclRuntime || ruleApplicationFailure != nil
|
||||
else {
|
||||
ruleApplicationFailure = nil
|
||||
return
|
||||
}
|
||||
modeChangeGeneration += 1
|
||||
var ticket = convergenceTicket()
|
||||
let result = await apply(
|
||||
nextDocument, ticket: &ticket,
|
||||
requiresReceipt: true, convergeProxyOnSuccess: false,
|
||||
preserveProxyOnFailure: true, proxyTail: .converge, rollback: plan)
|
||||
switch result {
|
||||
case .applied, .unchanged:
|
||||
if revision == ruleDocuments.revision { ruleApplicationFailure = nil }
|
||||
case .superseded:
|
||||
break
|
||||
case .failed:
|
||||
let failure = ProxyRuntimeFacts(state: state).failure ?? .service(.runtimeFile)
|
||||
_ = await restore(plan, ticket: &ticket)
|
||||
if convergenceIsCurrent(ticket), revision == ruleDocuments.revision {
|
||||
ruleApplicationFailure = failure
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 保存前校验(issue #66 AC2):固定本地冲突与重复整批拒绝并返回可解释原因。
|
||||
/// 同行动覆盖和相反行动遮蔽不阻止保存;编译时按当前骨架解释完整集合。
|
||||
func validateCustomRulesForPersistence(_ rules: [CustomRule]) -> (
|
||||
|
||||
@@ -17,21 +17,6 @@ extension ProxyRuntimeController {
|
||||
await transitionMode(proxyMode, ruleDefaultAction: action)
|
||||
}
|
||||
|
||||
/// 模式切换前的可恢复快照:失败时按它原样还原。
|
||||
private struct ModeTransitionSnapshot {
|
||||
let settings: ProxySettings
|
||||
let mode: ProxyMode
|
||||
/// 切换前的运行时文档;无活动文档时为 `nil`,恢复前由调用方补上当次加载值。
|
||||
let document: SslocalRuntimeDocument?
|
||||
let state: AgentRunState
|
||||
|
||||
/// 用当次加载的文档补齐快照(原 `previousDocument ?? currentDocument` 语义)。
|
||||
func resolvingDocument(_ fallback: SslocalRuntimeDocument) -> ModeTransitionSnapshot {
|
||||
ModeTransitionSnapshot(
|
||||
settings: settings, mode: mode, document: document ?? fallback, state: state)
|
||||
}
|
||||
}
|
||||
|
||||
private func transitionMode(
|
||||
_ mode: ProxyMode,
|
||||
ruleDefaultAction: RuleDefaultAction
|
||||
@@ -45,150 +30,6 @@ extension ProxyRuntimeController {
|
||||
settings = next
|
||||
proxyMode = mode
|
||||
modeChangeGeneration += 1
|
||||
var ticket = convergenceTicket()
|
||||
|
||||
guard settings.agentEnabled, state != .off,
|
||||
let currentDocument = lastDocument ?? runtimeFileStore.loadDocument()
|
||||
else {
|
||||
systemProxyObserver.modeTransitionAwaitingRuntime()
|
||||
return
|
||||
}
|
||||
|
||||
let nextDocument: SslocalRuntimeDocument
|
||||
do {
|
||||
nextDocument = try await runtimeDocument(currentDocument, for: mode)
|
||||
} catch RulePreparationError.superseded {
|
||||
return
|
||||
} catch {
|
||||
// 快照缺失/损坏:不静默退化,恢复旧模式与旧子选项。
|
||||
RuntimeLog.emit(.runtimePersistFailed(detail: String(describing: error)))
|
||||
ticket.preparation = runtimePreparationGeneration
|
||||
await restoreModeTransition(
|
||||
snapshot: snapshot.resolvingDocument(currentDocument), ticket: ticket)
|
||||
return
|
||||
}
|
||||
// 派生自身推进了派生代际:票据以派生后的当前值推进 preparation 分量。
|
||||
ticket.preparation = runtimePreparationGeneration
|
||||
guard nextDocument.aclRuntime != currentDocument.aclRuntime else {
|
||||
guard settings.systemProxyEnabled else { return }
|
||||
state = .starting
|
||||
_ = await presentLaunchHealth(nextDocument)
|
||||
return
|
||||
}
|
||||
|
||||
await deployModeTransition(
|
||||
nextDocument, snapshot: snapshot.resolvingDocument(currentDocument), ticket: ticket)
|
||||
}
|
||||
|
||||
enum RulePreparationError: Error { case superseded }
|
||||
|
||||
private struct RulePreparationTicket {
|
||||
let generation: Int
|
||||
let flow: Int
|
||||
let modeGeneration: Int
|
||||
let settings: ProxySettings
|
||||
let target: NodeID?
|
||||
let ruleRevision: Int?
|
||||
}
|
||||
|
||||
/// Capture mutable facts before leaving the UI actor. A stale task may not
|
||||
/// mutate runtime state or write a contract, even when its calculation fails.
|
||||
func runtimeDocument(
|
||||
_ source: SslocalRuntimeDocument, for mode: ProxyMode
|
||||
) async throws -> SslocalRuntimeDocument {
|
||||
runtimePreparationGeneration += 1
|
||||
let ticket = RulePreparationTicket(
|
||||
generation: runtimePreparationGeneration, flow: flowGeneration,
|
||||
modeGeneration: modeChangeGeneration, settings: settings, target: activeTargetID,
|
||||
ruleRevision: mode == .rule ? ruleDocuments.revision : nil)
|
||||
do {
|
||||
let user = mode == .rule ? try ruleDocuments.load() : nil
|
||||
let builtIn =
|
||||
mode == .rule ? try await runtimeBuiltinRules(ticket.settings.ruleDefaultAction) : []
|
||||
let aclURL = runtimeFileStore.aclFileURL
|
||||
let result = await Task.detached(priority: .userInitiated) {
|
||||
RuleRuntimeCompiler.compile(
|
||||
.init(
|
||||
source: source, mode: mode, defaultAction: ticket.settings.ruleDefaultAction,
|
||||
document: user, builtIn: builtIn, aclURL: aclURL))
|
||||
}.value
|
||||
guard preparationIsCurrent(ticket) else { throw RulePreparationError.superseded }
|
||||
return result
|
||||
} catch {
|
||||
guard preparationIsCurrent(ticket) else { throw RulePreparationError.superseded }
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
private func preparationIsCurrent(_ ticket: RulePreparationTicket) -> Bool {
|
||||
ticket.generation == runtimePreparationGeneration && ticket.flow == flowGeneration
|
||||
&& ticket.modeGeneration == modeChangeGeneration
|
||||
&& ticket.settings.listen == settings.listen
|
||||
&& ticket.settings.preferredMode == settings.preferredMode
|
||||
&& ticket.settings.ruleDefaultAction == settings.ruleDefaultAction
|
||||
&& ticket.settings.agentEnabled == settings.agentEnabled
|
||||
&& ticket.target == activeTargetID
|
||||
&& (ticket.ruleRevision == nil || ticket.ruleRevision == ruleDocuments.revision)
|
||||
}
|
||||
|
||||
private func runtimeBuiltinRules(_ action: RuleDefaultAction) async throws -> [ProxyRule] {
|
||||
switch action {
|
||||
case .proxyWhenUnmatched:
|
||||
let geolocation = try await ruleSnapshots.load(.geolocationCN)
|
||||
let china = try await ruleSnapshots.load(.chinaIPv4)
|
||||
return await Task.detached(priority: .userInitiated) {
|
||||
BuiltinRuleCatalog.chinaDirectRules(from: [geolocation, china])
|
||||
}.value
|
||||
case .directWhenUnmatched:
|
||||
let snapshot = try await ruleSnapshots.load(.gfwlist)
|
||||
return await Task.detached(priority: .userInitiated) {
|
||||
BuiltinRuleCatalog.gfwlistRules(from: snapshot)
|
||||
}.value
|
||||
}
|
||||
}
|
||||
|
||||
private func deployModeTransition(
|
||||
_ document: SslocalRuntimeDocument,
|
||||
snapshot: ModeTransitionSnapshot,
|
||||
ticket: ConvergenceTicket
|
||||
) async {
|
||||
var ticket = ticket
|
||||
// 两个调用点都经 resolvingDocument 补齐文档;防御性解包失败即无事可做。
|
||||
guard let rollbackDocument = snapshot.document else { return }
|
||||
let plan = RollbackPlan(
|
||||
document: rollbackDocument, state: snapshot.state,
|
||||
payload: .modeTransition(
|
||||
mode: snapshot.mode, ruleDefaultAction: snapshot.settings.ruleDefaultAction),
|
||||
systemProxyIntentAtCapture: snapshot.settings.systemProxyEnabled,
|
||||
convergeProxyOnIntentChange: true)
|
||||
let result = await apply(
|
||||
document, ticket: &ticket,
|
||||
requiresReceipt: true,
|
||||
convergeProxyOnSuccess: false,
|
||||
preserveProxyOnFailure: true,
|
||||
proxyTail: .forceApply, rollback: plan)
|
||||
if case .failed = result {
|
||||
await restoreModeTransition(snapshot: snapshot, ticket: ticket)
|
||||
}
|
||||
}
|
||||
|
||||
/// mode 切换的失败回滚:按快照整体恢复;载荷持久化失败以 serviceFailed
|
||||
/// 呈现(仅在旧运行时恢复到健康时)。
|
||||
private func restoreModeTransition(
|
||||
snapshot: ModeTransitionSnapshot,
|
||||
ticket: ConvergenceTicket
|
||||
) async {
|
||||
var ticket = ticket
|
||||
guard let previousDocument = snapshot.document else { return }
|
||||
let plan = RollbackPlan(
|
||||
document: previousDocument, state: snapshot.state,
|
||||
payload: .modeTransition(
|
||||
mode: snapshot.mode, ruleDefaultAction: snapshot.settings.ruleDefaultAction),
|
||||
systemProxyIntentAtCapture: snapshot.settings.systemProxyEnabled,
|
||||
convergeProxyOnIntentChange: true)
|
||||
let report = await restore(plan, ticket: &ticket)
|
||||
if report.runtimeHealthy == true, report.payloadFailureDescription != nil {
|
||||
state = .serviceFailed(.persistence)
|
||||
}
|
||||
await convergeRuntime(.modeTransition(snapshot))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,72 @@
|
||||
import Foundation
|
||||
|
||||
// 运行状态收敛 module 的配置派生实现;异步计算前捕获事实,恢复后拒绝旧结果。
|
||||
extension ProxyRuntimeController {
|
||||
enum RulePreparationError: Error { case superseded }
|
||||
|
||||
private struct RulePreparationTicket {
|
||||
let generation: Int
|
||||
let flow: Int
|
||||
let modeGeneration: Int
|
||||
let settings: ProxySettings
|
||||
let target: NodeID?
|
||||
let ruleRevision: Int?
|
||||
}
|
||||
|
||||
/// Capture mutable facts before leaving the UI actor. A stale task may not
|
||||
/// mutate runtime state or write a contract, even when its calculation fails.
|
||||
func runtimeDocument(
|
||||
_ source: SslocalRuntimeDocument, for mode: ProxyMode
|
||||
) async throws -> SslocalRuntimeDocument {
|
||||
runtimePreparationGeneration += 1
|
||||
let ticket = RulePreparationTicket(
|
||||
generation: runtimePreparationGeneration, flow: flowGeneration,
|
||||
modeGeneration: modeChangeGeneration, settings: settings, target: activeTargetID,
|
||||
ruleRevision: mode == .rule ? ruleDocuments.revision : nil)
|
||||
do {
|
||||
let user = mode == .rule ? try ruleDocuments.load() : nil
|
||||
let builtIn =
|
||||
mode == .rule ? try await runtimeBuiltinRules(ticket.settings.ruleDefaultAction) : []
|
||||
let aclURL = runtimeFileStore.aclFileURL
|
||||
let result = await Task.detached(priority: .userInitiated) {
|
||||
RuleRuntimeCompiler.compile(
|
||||
.init(
|
||||
source: source, mode: mode, defaultAction: ticket.settings.ruleDefaultAction,
|
||||
document: user, builtIn: builtIn, aclURL: aclURL))
|
||||
}.value
|
||||
guard preparationIsCurrent(ticket) else { throw RulePreparationError.superseded }
|
||||
return result
|
||||
} catch {
|
||||
guard preparationIsCurrent(ticket) else { throw RulePreparationError.superseded }
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
private func preparationIsCurrent(_ ticket: RulePreparationTicket) -> Bool {
|
||||
ticket.generation == runtimePreparationGeneration && ticket.flow == flowGeneration
|
||||
&& ticket.modeGeneration == modeChangeGeneration
|
||||
&& ticket.settings.listen == settings.listen
|
||||
&& ticket.settings.preferredMode == settings.preferredMode
|
||||
&& ticket.settings.ruleDefaultAction == settings.ruleDefaultAction
|
||||
&& ticket.settings.agentEnabled == settings.agentEnabled
|
||||
&& ticket.target == activeTargetID
|
||||
&& (ticket.ruleRevision == nil || ticket.ruleRevision == ruleDocuments.revision)
|
||||
}
|
||||
|
||||
private func runtimeBuiltinRules(_ action: RuleDefaultAction) async throws -> [ProxyRule] {
|
||||
switch action {
|
||||
case .proxyWhenUnmatched:
|
||||
let geolocation = try await ruleSnapshots.load(.geolocationCN)
|
||||
let china = try await ruleSnapshots.load(.chinaIPv4)
|
||||
return await Task.detached(priority: .userInitiated) {
|
||||
BuiltinRuleCatalog.chinaDirectRules(from: [geolocation, china])
|
||||
}.value
|
||||
case .directWhenUnmatched:
|
||||
let snapshot = try await ruleSnapshots.load(.gfwlist)
|
||||
return await Task.detached(priority: .userInitiated) {
|
||||
BuiltinRuleCatalog.gfwlistRules(from: snapshot)
|
||||
}.value
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -104,59 +104,7 @@ extension ProxyRuntimeControllerModeTests {
|
||||
XCTAssertEqual(controller.settings.preferredMode, ProxyModeKind.rule, "切换意图保留")
|
||||
}
|
||||
|
||||
/// restore 的时效门禁先于任何事实变更:agent 意图关闭或票据过期时弃权——
|
||||
/// 不执行计划、不重注册,回滚载荷不得写回持久层(否则会静默撤销用户的
|
||||
/// mode 意图)。
|
||||
func testRestoreSkipsPayloadPersistenceWhenAgentOffOrSuperseded() async throws {
|
||||
let seeded = try makeSeededCatalog()
|
||||
let settingsStore = InMemoryProxySettingsStore()
|
||||
let controller = makeController(
|
||||
probe: ProxyRuntimeFixture.FakeProbe.reachable(),
|
||||
settingsStore: settingsStore,
|
||||
settings: ProxySettings(
|
||||
listen: ActivationFixture.listen, preferredMode: .global, agentEnabled: true),
|
||||
proxyMode: .global)
|
||||
try await controller.activate(seeded.server)
|
||||
let document = try XCTUnwrap(controller.lastDocument)
|
||||
|
||||
// (a) agent 意图已关闭:门禁弃权。
|
||||
await controller.setAgentEnabled(false)
|
||||
let registrationsAfterStop = agent.registerCount
|
||||
var offTicket = controller.convergenceTicket()
|
||||
let offReport = await controller.restore(
|
||||
RollbackPlan(
|
||||
document: document, state: .running,
|
||||
payload: .modeTransition(mode: .rule, ruleDefaultAction: .proxyWhenUnmatched),
|
||||
systemProxyIntentAtCapture: false, convergeProxyOnIntentChange: false),
|
||||
ticket: &offTicket)
|
||||
XCTAssertNil(offReport.runtimeHealthy)
|
||||
XCTAssertNil(offReport.payloadFailureDescription)
|
||||
XCTAssertEqual(agent.registerCount, registrationsAfterStop, "弃权的回滚不得拉起旧运行时")
|
||||
XCTAssertEqual(settingsStore.saved?.preferredMode, ProxyModeKind.global, "旧载荷未写回持久层")
|
||||
XCTAssertFalse(settingsStore.saved?.agentEnabled ?? true)
|
||||
XCTAssertEqual(controller.state, .off)
|
||||
|
||||
// (b) 票据过期(mode 代际已前进):同样弃权且不写回。
|
||||
await controller.setAgentEnabled(true)
|
||||
await controller.setProxyMode(.direct)
|
||||
XCTAssertEqual(controller.state, .running)
|
||||
let staleTicket = controller.convergenceTicket()
|
||||
await controller.setProxyMode(.rule)
|
||||
XCTAssertEqual(controller.state, .running)
|
||||
let registrationsBeforeRestore = agent.registerCount
|
||||
var ticket = staleTicket
|
||||
let staleReport = await controller.restore(
|
||||
RollbackPlan(
|
||||
document: try XCTUnwrap(controller.lastDocument), state: .running,
|
||||
payload: .modeTransition(mode: .global, ruleDefaultAction: .proxyWhenUnmatched),
|
||||
systemProxyIntentAtCapture: false, convergeProxyOnIntentChange: false),
|
||||
ticket: &ticket)
|
||||
XCTAssertNil(staleReport.runtimeHealthy)
|
||||
XCTAssertEqual(agent.registerCount, registrationsBeforeRestore)
|
||||
XCTAssertEqual(settingsStore.saved?.preferredMode, ProxyModeKind.rule, "过期回滚不得改写持久层")
|
||||
}
|
||||
|
||||
/// agent-off 落在目录部署(deployPrepared 路径)的健康门窗口内:结果必须是
|
||||
/// agent-off 落在目录部署的健康门窗口内:结果必须是
|
||||
/// superseded(而非误分类的 failed),运行时保持 off、lastDocument 不复活。
|
||||
func testAgentOffDuringDeploymentHealthGateSupersedesDeployment() async throws {
|
||||
let seeded = try makeSeededCatalog()
|
||||
|
||||
@@ -0,0 +1,145 @@
|
||||
import Foundation
|
||||
import Testing
|
||||
|
||||
@testable import ShadowsocksX_NG2
|
||||
|
||||
/// 通过模式命令验证配置派生失败和无 ACL 变化路径,不调用恢复实现。
|
||||
@MainActor
|
||||
struct RuntimeConvergenceTests {
|
||||
@Test(arguments: [false, true])
|
||||
func modeDerivationFailureRestoresRuntimeAndReportsPersistenceFailure(
|
||||
restorePersistenceFails: Bool
|
||||
) async throws {
|
||||
let fixture = try Fixture(
|
||||
systemProxyEnabled: true,
|
||||
snapshots: BuiltinRuleSnapshots(loader: { _ in throw RuleSnapshotError.missing }))
|
||||
defer { fixture.cleanUp() }
|
||||
try await fixture.controller.activate(fixture.server)
|
||||
let previous = try #require(fixture.runtimeStore.loadDocument())
|
||||
let appliedCount = fixture.proxy.applied.count
|
||||
fixture.settings.rejectGlobalMode = restorePersistenceFails
|
||||
|
||||
await fixture.controller.setProxyMode(.rule)
|
||||
|
||||
#expect(fixture.controller.proxyMode == .global)
|
||||
#expect(fixture.controller.settings.preferredMode == .global)
|
||||
#expect(fixture.runtimeStore.loadDocument() == previous)
|
||||
#expect(
|
||||
fixture.controller.state
|
||||
== (restorePersistenceFails ? .serviceFailed(.persistence) : .running))
|
||||
#expect(fixture.settings.saved?.preferredMode == (restorePersistenceFails ? .rule : .global))
|
||||
#expect(fixture.controller.systemProxyState == .applied)
|
||||
#expect(fixture.proxy.applied.count == appliedCount)
|
||||
#expect(fixture.proxy.clearCount == 0)
|
||||
}
|
||||
|
||||
@Test(arguments: [false, true])
|
||||
func unchangedModeACLOnlyChecksHealthWhenSystemProxyIsEnabled(
|
||||
systemProxyEnabled: Bool
|
||||
) async throws {
|
||||
let fixture = try Fixture(systemProxyEnabled: systemProxyEnabled)
|
||||
defer { fixture.cleanUp() }
|
||||
try await fixture.controller.activate(fixture.server)
|
||||
let bytes = try Data(contentsOf: fixture.runtimeStore.fileURL)
|
||||
let registers = fixture.agent.registerCount
|
||||
let unregisters = fixture.agent.unregisterCount
|
||||
let probes = fixture.probe.ports.count
|
||||
let signals = fixture.signals.signalsSent.filter { $0.signal != 0 }.count
|
||||
|
||||
// 全局模式仍保存规则默认动作,但它不改变全局 ACL。
|
||||
await fixture.controller.setRuleDefaultAction(.directWhenUnmatched)
|
||||
|
||||
#expect(fixture.settings.saved?.ruleDefaultAction == .directWhenUnmatched)
|
||||
#expect(fixture.controller.proxyMode == .global)
|
||||
#expect(fixture.controller.state == .running)
|
||||
#expect(try Data(contentsOf: fixture.runtimeStore.fileURL) == bytes)
|
||||
#expect(fixture.agent.registerCount == registers)
|
||||
#expect(fixture.agent.unregisterCount == unregisters)
|
||||
#expect(fixture.signals.signalsSent.filter { $0.signal != 0 }.count == signals)
|
||||
#expect(fixture.probe.ports.count == probes + (systemProxyEnabled ? 2 : 0))
|
||||
}
|
||||
|
||||
@MainActor
|
||||
private struct Fixture {
|
||||
let runtime = ProxyRuntimeFixture.makeTemporaryRuntime()
|
||||
let agent = ProxyRuntimeFixture.FakeLaunchAgent()
|
||||
let probe = ProxyRuntimeFixture.FakeProbe.reachable()
|
||||
let proxy = ProxyRuntimeFixture.FakeSystemProxy()
|
||||
let settings = ModeRestoreSettingsStore()
|
||||
let signals = ProxyRuntimeControllerTests.SignalRecorder()
|
||||
let runtimeStore: RuntimeFileStore
|
||||
let controller: ProxyRuntimeController
|
||||
let server: NodeID
|
||||
|
||||
init(
|
||||
systemProxyEnabled: Bool,
|
||||
snapshots: BuiltinRuleSnapshots = BuiltinRuleSnapshots(
|
||||
loader: ProxyRuntimeFixture.controlFlowRuleSnapshot)
|
||||
) throws {
|
||||
let credentials = InMemoryCredentialStore()
|
||||
var catalog = ConfigurationCatalog()
|
||||
server = try ActivationFixture.addPlainServer("收敛测试", in: &catalog, credentials: credentials)
|
||||
let catalogURL = runtime.directory.appendingPathComponent("catalog.json")
|
||||
try CatalogFileStore(fileURL: catalogURL).save(CatalogDocument(catalog: catalog))
|
||||
runtimeStore = RuntimeFileStore(fileURL: runtime.contract)
|
||||
let store = runtimeStore
|
||||
let recorder = signals
|
||||
agent.onUnregister = { recorder.terminateWrapper() }
|
||||
agent.onRegister = {
|
||||
recorder.relaunchWrapper()
|
||||
if let document = store.loadDocument() {
|
||||
try? store.writeRuntimeReceipt(for: document, processID: 42)
|
||||
}
|
||||
}
|
||||
recorder.reloadReceipt = {
|
||||
if let document = store.loadDocument() {
|
||||
try? store.writeRuntimeReceipt(for: document, processID: 42)
|
||||
}
|
||||
}
|
||||
controller = ProxyRuntimeController(
|
||||
catalogSnapshotReader: ProxyRuntimeFixture.catalogSnapshotReader(at: catalogURL),
|
||||
activationFileStore: ActivationStateFileStore(
|
||||
fileURL: runtime.directory.appendingPathComponent("activation.json")),
|
||||
runtimeFileStore: runtimeStore, credentials: credentials,
|
||||
plugins: ActivationFixture.plugins, settingsStore: settings,
|
||||
customRuleStore: CustomRuleStore(
|
||||
fileURL: runtime.directory.appendingPathComponent("rules.json")),
|
||||
appBundle: AppArtifact.bundle, ruleSnapshots: snapshots,
|
||||
settingsRestore: RestoredProxySettings(
|
||||
settings: ProxySettings(
|
||||
listen: ActivationFixture.listen, preferredMode: .global,
|
||||
agentEnabled: true, systemProxyEnabled: systemProxyEnabled),
|
||||
unreadableError: nil),
|
||||
agent: agent, probe: probe, systemProxy: proxy,
|
||||
systemProxyHelper: ProxyRuntimeFixture.FakeSystemProxyHelperService(),
|
||||
proxyMode: .global, firewallChecker: ProxyRuntimeFixture.FakeFirewallChecker(),
|
||||
launchHealthTimeoutSeconds: 0.05,
|
||||
launchHealthRetryDelay: { await Task.yield() },
|
||||
systemProxyHealthPollIntervalNanoseconds: 60_000_000_000,
|
||||
helperRefreshDelayNanoseconds: 0,
|
||||
sendSignal: { recorder.send($0, $1) },
|
||||
processIsAlive: { recorder.send($0, 0) == 0 })
|
||||
}
|
||||
|
||||
func cleanUp() {
|
||||
controller.systemProxyObserver.systemProxyHealthTask?.cancel()
|
||||
controller.cancelFirewallObservation()
|
||||
try? FileManager.default.removeItem(at: runtime.directory)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private final class ModeRestoreSettingsStore: ProxySettingsStoring {
|
||||
private let store = InMemoryProxySettingsStore()
|
||||
var rejectGlobalMode = false
|
||||
var saved: ProxySettings? { store.saved }
|
||||
|
||||
func load() throws -> ProxySettings { try store.load() }
|
||||
|
||||
func save(_ settings: ProxySettings) throws {
|
||||
if rejectGlobalMode && settings.preferredMode == .global {
|
||||
throw ProxySettingsStoreError.ioFailure(detail: "restore fixture failure")
|
||||
}
|
||||
try store.save(settings)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user