mirror of
https://github.com/shadowsocks/ShadowsocksX-NG.git
synced 2026-11-08 11:14:51 +00:00
401 lines
16 KiB
Swift
401 lines
16 KiB
Swift
import Foundation
|
|
|
|
/// 一次计划执行的结果:动作是否全部按序成功,以及本次执行占用的并发流
|
|
/// 代际;收敛票据用它推进 flow,避免调用方手算 `flowGeneration + 1`。
|
|
struct RuntimeExecutionOutcome {
|
|
/// 本次执行占用的 `flowGeneration`。
|
|
let flow: Int
|
|
/// 全部计划动作是否按序成功。
|
|
let succeeded: Bool
|
|
}
|
|
|
|
/// 收敛票据:运行时收敛操作的代际快照。入口捕获三个并发代际,之后每步
|
|
/// 推进用它判定「从入口到现在,并发流 / 模式切换 / 运行时派生是否前进了,
|
|
/// 会话内 agent 意图是否仍开启」;被更新提交或新意图取代的旧收敛不得再
|
|
/// 变更运行时状态或写契约。
|
|
///
|
|
/// 四项失效事实由票据恒全查,不随调用路径增减——「什么会使一次收敛失效」
|
|
/// 是票据的固有语义(逐路径手选子集曾让 agent 关闭在 mode 切换的健康门
|
|
/// await 窗口内被误分类为失败并触发复活性的 restore)。各路径只负责捕获
|
|
/// 与推进时机:flow 分量在 execute 后以实际占用值推进,mode 切换路径在
|
|
/// 派生后推进 preparation 分量(派生自身会推进它),rules 部署在推进 mode
|
|
/// 代际之后捕获。
|
|
private struct ConvergenceTicket: Equatable {
|
|
/// 并发流代际;每次 execute 之后由调用方以实际占用值推进。
|
|
var flow: Int
|
|
/// 模式切换代际;捕获后不再变化。
|
|
let mode: Int
|
|
/// 运行时文档派生代际;每次派生之后由调用方以当前值推进。
|
|
var preparation: Int
|
|
}
|
|
|
|
extension ProxyRuntimeController {
|
|
/// 以当前代际捕获收敛票据。
|
|
private func convergenceTicket() -> ConvergenceTicket {
|
|
ConvergenceTicket(
|
|
flow: flowGeneration, mode: modeChangeGeneration,
|
|
preparation: runtimePreparationGeneration)
|
|
}
|
|
|
|
/// 票据是否仍代表当前收敛:并发流 / 模式切换 / 派生三个代际均未前进,
|
|
/// 且会话内 agent 意图仍开启。
|
|
private func convergenceIsCurrent(_ ticket: ConvergenceTicket) -> Bool {
|
|
ticket.flow == flowGeneration && ticket.mode == modeChangeGeneration
|
|
&& ticket.preparation == runtimePreparationGeneration
|
|
&& settings.agentEnabled
|
|
}
|
|
|
|
/// 派生文档对当前运行时是否无事可做（GLOSSARY.md「有效值未变的提交不做
|
|
/// 运行时收敛」）：发布意图相同、输出字节未漂移、agent 已注册且 wrapper
|
|
/// 存活、控制器处于健康运行态——计划层动作序列为空即幂等跳过。判定在
|
|
/// 置 `starting` 之前进行，跳过路径不闪状态、不重走健康门；磁盘漂移或
|
|
/// wrapper 失踪时动作序列自然非空，走完整 deploy 自愈。
|
|
private func convergencePlanIsEmpty(
|
|
_ document: SslocalRuntimeDocument, preparedContract: PreparedRuntimeContract?
|
|
) -> Bool {
|
|
guard runtimeReloadProcesses.isEmpty else { return false }
|
|
switch state {
|
|
case .running, .firewallBlocked:
|
|
break
|
|
case .off, .starting, .launchFailed, .requiresApproval, .serviceFailed:
|
|
return false
|
|
}
|
|
let actions = ProxyRuntimePlan.actions(
|
|
intent: .run(document),
|
|
agentStatus: agent.status,
|
|
wrapper: wrapperState(),
|
|
configurationIsCurrent: preparedContract.map { runtimeFileStore.contains($0) } ?? false)
|
|
return actions.isEmpty
|
|
}
|
|
|
|
/// reexpand 三-case 开关的收敛主体（显式收敛、设置保存、启动重同步共用）：
|
|
/// deployed → 部署派生文档；目标失效 → 清除并按意图收敛；无目标 → 空监听。
|
|
/// 幂等跳过由 apply 层的 plan 空判定承担；目录重校验（跳过名单、目标失效
|
|
/// 清除）已在 `reexpand` 完成，不受其影响。
|
|
func convergeToReexpanded(_ catalog: ConfigurationCatalog) async {
|
|
guard !settingsUnreadable else {
|
|
state = .launchFailed(.unreadableSettings)
|
|
return
|
|
}
|
|
guard catalogSnapshotReader.isCatalogAvailable else { return }
|
|
switch reexpand(in: catalog) {
|
|
case .deployed(let configuration):
|
|
await deploy(configuration.document)
|
|
case .clearedAndStopped(let failure):
|
|
await handleCleared(failure)
|
|
case nil:
|
|
await deployListeningWithoutTarget()
|
|
}
|
|
}
|
|
|
|
/// 一个意图入口隐藏派生、执行政策和失败呈现；调用者不组合执行选项。
|
|
enum RuntimeConvergenceIntent {
|
|
case deployment(SslocalRuntimeDocument)
|
|
case modeTransition
|
|
case savedRules
|
|
}
|
|
|
|
enum ConvergenceResult {
|
|
case applied
|
|
case unchanged
|
|
/// 派生或部署失败；失败呈现已经处理。
|
|
case failed
|
|
/// 被新意图取代，不得发布旧结果。
|
|
case superseded
|
|
}
|
|
|
|
@discardableResult
|
|
func convergeRuntime(_ intent: RuntimeConvergenceIntent) async -> ConvergenceResult {
|
|
guard !settingsUnreadable else {
|
|
state = .launchFailed(.unreadableSettings)
|
|
return .failed
|
|
}
|
|
guard catalogSnapshotReader.isCatalogAvailable else { return .failed }
|
|
switch intent {
|
|
case .deployment(let source): return await convergeDeployment(source)
|
|
case .modeTransition: return await convergeModeTransition()
|
|
case .savedRules: return await convergeSavedRules()
|
|
}
|
|
}
|
|
|
|
private func convergeDeployment(_ source: SslocalRuntimeDocument) async -> ConvergenceResult {
|
|
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() async -> ConvergenceResult {
|
|
await convergeChangedIntent(policy: .modeTransition)
|
|
}
|
|
|
|
private func convergeSavedRules() async -> ConvergenceResult {
|
|
guard proxyMode == .rule else { return .unchanged }
|
|
let revision = ruleDocuments.revision
|
|
let result = await convergeChangedIntent(policy: .savedRules)
|
|
switch result {
|
|
case .applied, .unchanged:
|
|
if revision == ruleDocuments.revision { ruleApplicationFailure = nil }
|
|
case .superseded:
|
|
break
|
|
case .failed:
|
|
if revision == ruleDocuments.revision {
|
|
ruleApplicationFailure = ProxyRuntimeFacts(state: state).failure ?? .service(.runtimeFile)
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
|
|
/// 三种意图的执行差异只在 module 内选择，不能由调用者拼成任意组合。
|
|
private enum DeploymentPolicy {
|
|
case deployment
|
|
case modeTransition
|
|
case savedRules
|
|
}
|
|
|
|
private enum ChangedIntentSource {
|
|
case ready(SslocalRuntimeDocument)
|
|
case completed(ConvergenceResult)
|
|
}
|
|
|
|
private struct PreparedChangedIntent {
|
|
let document: SslocalRuntimeDocument
|
|
let contract: PreparedRuntimeContract
|
|
let ticket: ConvergenceTicket
|
|
}
|
|
|
|
private enum ChangedIntentPreparation {
|
|
case ready(PreparedChangedIntent)
|
|
case completed(ConvergenceResult)
|
|
}
|
|
|
|
private struct RuntimeApplyAttempt {
|
|
let document: SslocalRuntimeDocument
|
|
let contract: PreparedRuntimeContract?
|
|
let policy: DeploymentPolicy
|
|
let reloadingProcessID: Int32?
|
|
let reloadID: UUID
|
|
}
|
|
|
|
private enum ApplyPreparation {
|
|
case ready(RuntimeApplyAttempt)
|
|
case completed(ConvergenceResult)
|
|
}
|
|
|
|
/// 模式与规则变更从同一份已保存资料重新展开，不以输出配置作为派生输入。
|
|
private func convergeChangedIntent(policy: DeploymentPolicy) async -> ConvergenceResult {
|
|
switch await changedIntentSource(policy) {
|
|
case .completed(let result): return result
|
|
case .ready(let source): return await prepareAndConverge(source, policy: policy)
|
|
}
|
|
}
|
|
|
|
private func changedIntentSource(_ policy: DeploymentPolicy) async -> ChangedIntentSource {
|
|
guard settings.agentEnabled, state != .off else {
|
|
if policy == .modeTransition { systemProxyObserver.modeTransitionAwaitingRuntime() }
|
|
return .completed(.unchanged)
|
|
}
|
|
switch reexpand(in: catalogSnapshotReader.catalogSnapshot) {
|
|
case .deployed(let configuration): return .ready(configuration.document)
|
|
case .clearedAndStopped(let failure):
|
|
return .completed(await handleCleared(failure))
|
|
case nil: return .ready(SslocalRuntimeDocument(servers: [], listen: settings.listen))
|
|
}
|
|
}
|
|
|
|
private func prepareAndConverge(
|
|
_ source: SslocalRuntimeDocument, policy: DeploymentPolicy
|
|
) async -> ConvergenceResult {
|
|
let intent: PreparedChangedIntent
|
|
switch await prepareChangedIntent(source, policy: policy) {
|
|
case .completed(let result): return result
|
|
case .ready(let prepared): intent = prepared
|
|
}
|
|
if let unchanged = await unchangedIntentResult(intent, policy: policy) { return unchanged }
|
|
var ticket = intent.ticket
|
|
if policy == .savedRules {
|
|
modeChangeGeneration += 1
|
|
ticket = convergenceTicket()
|
|
}
|
|
return await apply(intent.document, ticket: &ticket, policy: policy)
|
|
}
|
|
|
|
private func prepareChangedIntent(
|
|
_ source: SslocalRuntimeDocument, policy: DeploymentPolicy
|
|
) async -> ChangedIntentPreparation {
|
|
var ticket = convergenceTicket()
|
|
let next: SslocalRuntimeDocument
|
|
let contract: PreparedRuntimeContract
|
|
do {
|
|
next = try await runtimeDocument(source, for: proxyMode)
|
|
contract = try PreparedRuntimeContract(next)
|
|
} catch RulePreparationError.superseded {
|
|
return .completed(.superseded)
|
|
} catch {
|
|
RuntimeLog.emit(.runtimePersistFailed(detail: String(describing: error)))
|
|
if policy == .savedRules {
|
|
ruleApplicationFailure = .service(.runtimeFile)
|
|
} else {
|
|
state = .serviceFailed(.runtimeFile)
|
|
}
|
|
return .completed(.failed)
|
|
}
|
|
ticket.preparation = runtimePreparationGeneration
|
|
return .ready(PreparedChangedIntent(document: next, contract: contract, ticket: ticket))
|
|
}
|
|
|
|
private func unchangedIntentResult(
|
|
_ intent: PreparedChangedIntent, policy: DeploymentPolicy
|
|
) async -> ConvergenceResult? {
|
|
let unchanged = runtimeFileStore.contains(intent.contract)
|
|
let hasNoDeploymentFailure = runtimeFileStore.readFailure() == nil
|
|
if policy == .savedRules, unchanged, ruleApplicationFailure == nil, hasNoDeploymentFailure {
|
|
return .unchanged
|
|
}
|
|
guard policy == .modeTransition, unchanged, hasNoDeploymentFailure,
|
|
runtimeIsAlreadyRunning, aclReloadProcessID(for: intent.document) != nil
|
|
else { return nil }
|
|
guard settings.systemProxyEnabled else { return .unchanged }
|
|
await systemProxyObserver.convergeSystemProxy()
|
|
return .applied
|
|
}
|
|
|
|
private var runtimeIsAlreadyRunning: Bool {
|
|
switch state {
|
|
case .running, .firewallBlocked: return true
|
|
default: return false
|
|
}
|
|
}
|
|
|
|
/// 仅 ACL 变化且原有子进程仍健康时，继续沿用监听事实。
|
|
private func aclReloadProcessID(for document: SslocalRuntimeDocument) -> Int32? {
|
|
switch state {
|
|
case .running, .firewallBlocked: break
|
|
default: return nil
|
|
}
|
|
guard agent.status == .registered, case .running = wrapperState(),
|
|
let identity = document.replacingACL(nil).intentSHA256,
|
|
runtimeObservation?.configurationIntentSHA256 == identity
|
|
else { return nil }
|
|
return effectiveRuntimeListenerProcessID
|
|
?? runtimeReloadProcesses.values.first(where: { processIsAlive($0) })
|
|
}
|
|
|
|
/// 共用执行脊柱；票据不越过意图 seam。
|
|
private func apply(
|
|
_ document: SslocalRuntimeDocument,
|
|
ticket: inout ConvergenceTicket,
|
|
policy: DeploymentPolicy
|
|
) async -> ConvergenceResult {
|
|
let preparation = await prepareApply(document, ticket: ticket, policy: policy)
|
|
let attempt: RuntimeApplyAttempt
|
|
switch preparation {
|
|
case .completed(let result): return result
|
|
case .ready(let prepared): attempt = prepared
|
|
}
|
|
defer { runtimeReloadProcesses.removeValue(forKey: attempt.reloadID) }
|
|
let execution = await execute(
|
|
.run(document), document: document, preparedContract: attempt.contract)
|
|
ticket.flow = execution.flow
|
|
return await completeApply(execution, attempt: attempt, ticket: &ticket)
|
|
}
|
|
|
|
private func prepareApply(
|
|
_ document: SslocalRuntimeDocument, ticket: ConvergenceTicket,
|
|
policy: DeploymentPolicy
|
|
) async -> ApplyPreparation {
|
|
let contract = try? PreparedRuntimeContract(document)
|
|
if runtimeFileStore.blocksDeployment(document) {
|
|
state = .serviceFailed(.agent)
|
|
await systemProxyObserver.holdSystemProxyIntent()
|
|
return .completed(.failed)
|
|
}
|
|
if convergencePlanIsEmpty(document, preparedContract: contract) {
|
|
RuntimeLog.emit(.contractUnchanged)
|
|
ruleApplicationFailure = nil
|
|
return .completed(.unchanged)
|
|
}
|
|
guard convergenceIsCurrent(ticket) else { return .completed(.superseded) }
|
|
let reloadingProcessID = aclReloadProcessID(for: document)
|
|
let reloadID = UUID()
|
|
if let reloadingProcessID { runtimeReloadProcesses[reloadID] = reloadingProcessID }
|
|
runtimeObservation = RuntimeDeploymentFacts(document)
|
|
if reloadingProcessID == nil { state = .starting }
|
|
return .ready(
|
|
RuntimeApplyAttempt(
|
|
document: document, contract: contract, policy: policy,
|
|
reloadingProcessID: reloadingProcessID, reloadID: reloadID))
|
|
}
|
|
|
|
private func completeApply(
|
|
_ execution: RuntimeExecutionOutcome, attempt: RuntimeApplyAttempt,
|
|
ticket: inout ConvergenceTicket
|
|
) async -> ConvergenceResult {
|
|
ticket.flow = execution.flow
|
|
guard convergenceIsCurrent(ticket) else { return .superseded }
|
|
guard execution.succeeded else {
|
|
if state != .requiresApproval,
|
|
runtimeFileStore.receiptIdentity(for: attempt.document) != nil
|
|
{
|
|
await failDeployment(ticket: &ticket)
|
|
}
|
|
return .failed
|
|
}
|
|
let healthy = await presentLaunchHealth(
|
|
attempt.document,
|
|
requiresReceipt: attempt.policy != .deployment || attempt.document.aclRuntime != nil,
|
|
convergeProxyOnSuccess: attempt.policy == .deployment,
|
|
preserveProxyOnFailure: false, preparedContract: attempt.contract,
|
|
reloadingProcessID: attempt.reloadingProcessID)
|
|
guard convergenceIsCurrent(ticket) else { return .superseded }
|
|
guard healthy else {
|
|
await failDeployment(ticket: &ticket)
|
|
return .failed
|
|
}
|
|
do { try runtimeFileStore.settleSuccessfulDeployment() } catch {
|
|
state = .serviceFailed(.runtimeFile)
|
|
await failDeployment(ticket: &ticket)
|
|
return .failed
|
|
}
|
|
runtimeObservation = RuntimeDeploymentFacts(attempt.document)
|
|
ruleApplicationFailure = nil
|
|
await convergeProxyAfterApply(attempt.policy)
|
|
return .applied
|
|
}
|
|
|
|
private func convergeProxyAfterApply(_ policy: DeploymentPolicy) async {
|
|
switch policy {
|
|
case .modeTransition: await systemProxyObserver.convergeSystemProxy(forceApply: true)
|
|
case .savedRules: await systemProxyObserver.convergeSystemProxy()
|
|
case .deployment: break
|
|
}
|
|
}
|
|
|
|
/// 发布后失败保留配置及失败身份，停止服务但不改写两个开关意图。
|
|
private func failDeployment(ticket: inout ConvergenceTicket) async {
|
|
guard convergenceIsCurrent(ticket) else { return }
|
|
var failureState = state
|
|
do { try runtimeFileStore.recordFailure(reason: "GUI deployment verification failed") } catch {
|
|
failureState = .serviceFailed(.runtimeFile)
|
|
}
|
|
let stopped = await unregisterFailedDeployment()
|
|
guard convergenceIsCurrent(ticket) else { return }
|
|
if stopped { state = failureState } else { state = .serviceFailed(.agent) }
|
|
await systemProxyObserver.holdSystemProxyIntent()
|
|
}
|
|
|
|
private func unregisterFailedDeployment() async -> Bool {
|
|
do { try agent.unregister() } catch { return false }
|
|
await waitForWrapperExit()
|
|
return wrapperState() == .notRunning && !runtimeFileStore.hasUnconfirmedChild
|
|
}
|
|
}
|