处理权限逻辑
This commit is contained in:
parent
3fa567762b
commit
acab387898
@ -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))
|
||||
|
||||
@ -9,54 +9,55 @@ import NIO
|
||||
|
||||
actor PolicyRuleStore {
|
||||
|
||||
// 处理权限的请求问题
|
||||
nonisolated private let cooldown: Duration = .seconds(5)
|
||||
// identityId
|
||||
private var coolingDown: Set<UInt32> = []
|
||||
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<PolicyRuleSnapshot>
|
||||
private var ruleMapByIdentity: [UInt32: PolicyRuleMap] = [:]
|
||||
private var policyByIdentity: [UInt32: PolicyEntry] = [:]
|
||||
|
||||
init(publisher: SnapshotPublisher<PolicyRuleSnapshot>) {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user