diff --git a/Tun/Context/SDLContextActor.swift b/Tun/Context/SDLContextActor.swift index 191e007..2ed6736 100644 --- a/Tun/Context/SDLContextActor.swift +++ b/Tun/Context/SDLContextActor.swift @@ -358,7 +358,7 @@ extension SDLContextActor { let updatePolicyWorker = PeriodicWorker( configuration: .init( - interval: .seconds(120), + interval: .seconds(10), runImmediately: true, mode: .fixedDelay, errorPolicy: .keepRunning(delay: .seconds(5)) diff --git a/Tun/Policy/PolicyRuleStore.swift b/Tun/Policy/PolicyRuleStore.swift index adc8661..4bac670 100644 --- a/Tun/Policy/PolicyRuleStore.swift +++ b/Tun/Policy/PolicyRuleStore.swift @@ -8,55 +8,56 @@ import Foundation import NIO actor PolicyRuleStore { - - // 处理权限的请求问题 - nonisolated private let cooldown: Duration = .seconds(5) - // identityId - private var coolingDown: Set = [] + + private struct PolicyEntry { + let ruleMap: PolicyRuleMap + let expiresAt: TimeInterval + } + + private struct PendingRequest { + let version: UInt32 + let retryCount: Int + let nextRetryAt: TimeInterval + } + + private let policyTTL: TimeInterval = 120 + private let refreshLeadTime: TimeInterval = 15 + private let baseRetryDelay: TimeInterval = 2 + private let maxRetryDelay: TimeInterval = 60 + // 处理各个请求的版本问题, map[identityId] = version private var versions: [UInt32: UInt32] = [:] + private var pendingByIdentity: [UInt32: PendingRequest] = [:] nonisolated private let alloctor = ByteBufferAllocator() private let publisher: SnapshotPublisher - private var ruleMapByIdentity: [UInt32: PolicyRuleMap] = [:] + private var policyByIdentity: [UInt32: PolicyEntry] = [:] init(publisher: SnapshotPublisher) { self.publisher = publisher } func makeBatchPolicyRequests(dstIdentityID: UInt32) -> [Data] { - return self.ruleMapByIdentity.keys.compactMap { identityId in - var policyRequest = SDLPolicyRequest() - policyRequest.srcIdentityID = identityId - policyRequest.dstIdentityID = dstIdentityID - policyRequest.version = self.nextVersion(identityId: identityId) - return try? policyRequest.serializedData() + let now = Date().timeIntervalSince1970 + let identities = Set(self.policyByIdentity.keys).union(self.pendingByIdentity.keys) + + return identities.compactMap { identityId in + self.makePolicyRequestIfDue(srcIdentityId: identityId, dstIdentityId: dstIdentityID, now: now, forceMissing: false) } } func makePolicyRequest(srcIdentityId: UInt32, dstIdentityId: UInt32) -> Data? { - guard self.ruleMapByIdentity[srcIdentityId] == nil else { + guard self.policyByIdentity[srcIdentityId] == nil else { return nil } - guard !coolingDown.contains(srcIdentityId) else { - return nil - } - - var policyRequest = SDLPolicyRequest() - policyRequest.srcIdentityID = srcIdentityId - policyRequest.dstIdentityID = dstIdentityId - policyRequest.version = self.nextVersion(identityId: srcIdentityId) - - coolingDown.insert(srcIdentityId) - - Task { [weak self] in - try? await Task.sleep(for: .seconds(5)) - await self?.endCooldown(for: srcIdentityId) - } - - return try? policyRequest.serializedData() + return self.makePolicyRequestIfDue( + srcIdentityId: srcIdentityId, + dstIdentityId: dstIdentityId, + now: Date().timeIntervalSince1970, + forceMissing: true + ) } // 处理权限的响应 @@ -64,7 +65,11 @@ actor PolicyRuleStore { let id = policyResponse.srcIdentityID let version = policyResponse.version - guard self.ruleMapByIdentity[id] == nil || ((self.ruleMapByIdentity[id]?.version ?? 0) < version) else { + if let pending = self.pendingByIdentity[id], pending.version <= version { + self.pendingByIdentity.removeValue(forKey: id) + } + + guard self.policyByIdentity[id] == nil || ((self.policyByIdentity[id]?.ruleMap.version ?? 0) < version) else { return } @@ -78,7 +83,11 @@ actor PolicyRuleStore { } ruleMap[proto, default: [:]][port] = true } - self.ruleMapByIdentity[id] = PolicyRuleMap(version: version, ruleMap: ruleMap) + let now = Date().timeIntervalSince1970 + self.policyByIdentity[id] = PolicyEntry( + ruleMap: PolicyRuleMap(version: version, ruleMap: ruleMap), + expiresAt: now + self.policyTTL + ) SDLLogger.log("[PolicyRuleStore] apply policy response, srcIdentityID: \(id), version: \(version), rulesCount: \(ruleMap.reduce(0) { $0 + $1.value.count })", for: .debug) // 发布新的快照信息 @@ -87,18 +96,14 @@ actor PolicyRuleStore { } func clear() { - self.coolingDown.removeAll() self.versions.removeAll() - self.ruleMapByIdentity.removeAll() + self.pendingByIdentity.removeAll() + self.policyByIdentity.removeAll() self.publisher.publish(PolicyRuleSnapshot.empty()) } private func compileSnapshot() -> PolicyRuleSnapshot { - return PolicyRuleSnapshot(ruleMapByIdentity: ruleMapByIdentity) - } - - private func endCooldown(for key: UInt32) { - self.coolingDown.remove(key) + return PolicyRuleSnapshot(ruleMapByIdentity: self.policyByIdentity.mapValues(\.ruleMap)) } private func nextVersion(identityId: UInt32) -> UInt32 { @@ -108,4 +113,46 @@ actor PolicyRuleStore { return version } + + private func makePolicyRequestIfDue(srcIdentityId: UInt32, dstIdentityId: UInt32, now: TimeInterval, forceMissing: Bool) -> Data? { + if let pending = self.pendingByIdentity[srcIdentityId], pending.nextRetryAt > now { + return nil + } + + if !forceMissing, + let entry = self.policyByIdentity[srcIdentityId], + entry.expiresAt - self.refreshLeadTime > now { + return nil + } + + let version = self.nextVersion(identityId: srcIdentityId) + guard let data = self.makePolicyRequestData(srcIdentityId: srcIdentityId, dstIdentityId: dstIdentityId, version: version) else { + return nil + } + + let retryCount = self.pendingByIdentity[srcIdentityId].map { $0.retryCount + 1 } ?? 0 + let retryDelay = self.retryDelay(for: retryCount, identityId: srcIdentityId) + self.pendingByIdentity[srcIdentityId] = PendingRequest( + version: version, + retryCount: retryCount, + nextRetryAt: now + retryDelay + ) + + return data + } + + private func makePolicyRequestData(srcIdentityId: UInt32, dstIdentityId: UInt32, version: UInt32) -> Data? { + var policyRequest = SDLPolicyRequest() + policyRequest.srcIdentityID = srcIdentityId + policyRequest.dstIdentityID = dstIdentityId + policyRequest.version = version + return try? policyRequest.serializedData() + } + + private func retryDelay(for retryCount: Int, identityId: UInt32) -> TimeInterval { + let cappedRetryCount = min(retryCount, 5) + let multiplier = Double(1 << cappedRetryCount) + let jitter = Double(identityId % 1_000) / 1_000 + return min(self.baseRetryDelay * multiplier + jitter, self.maxRetryDelay) + } }