fix:fix counter expired no recover bug. (#638)

Author: SkyeBeFreeman

polaris-ratelimit/polaris-ratelimit-client/src/main/java/com/tencent/polaris/ratelimit/client/flow/RateLimitWindow.java

@@ -93,6 +93,8 @@ public enum WindowStatus {
 
     private final AtomicLong lastAccessTimeMs = new AtomicLong();
 
+    private final AtomicLong lastInitTimeMs = new AtomicLong();
+
     // 执行正式分配的令牌桶
     private final QuotaBucket allocatingBucket;
 
@@ -314,6 +316,14 @@ public boolean isExpired() {
         return expired;
     }
 
+    public long getLastInitTimeMs() {
+        return lastInitTimeMs.get();
+    }
+
+    public void setLastInitTimeMs(long lastInitTimeMs) {
+        this.lastInitTimeMs.set(lastInitTimeMs);
+    }
+
     /**
      * 获取当前窗口的状态
      *

polaris-ratelimit/polaris-ratelimit-client/src/main/java/com/tencent/polaris/ratelimit/client/flow/StreamResource.java

@@ -142,6 +142,10 @@ public void closeStream(boolean closeSend) {
             if (null != channel) {
                 channel.shutdown();
             }
+            // 重置窗口最后初始化时间，清初始化窗口记录、上报索引记录
+            initRecord.forEach((serviceIdentifier, record) -> record.getRateLimitWindow().setLastInitTimeMs(0));
+            initRecord.clear();
+            counters.clear();
         }
     }
 
@@ -152,7 +156,9 @@ public void onNext(RateLimitResponse rateLimitResponse) {
         if (RateLimitCmd.INIT.equals(rateLimitResponse.getCmd())) {
             handleRateLimitInitResponse(rateLimitResponse.getRateLimitInitResponse());
         } else if (RateLimitCmd.ACQUIRE.equals(rateLimitResponse.getCmd())) {
-            handleRateLimitReportResponse(rateLimitResponse.getRateLimitReportResponse());
+            if (!handleRateLimitReportResponse(rateLimitResponse.getRateLimitReportResponse())) {
+                closeStream(true);
+            }
         }
     }
 
@@ -221,6 +227,7 @@ private void handleRateLimitInitResponse(RateLimitInitResponse rateLimitInitResp
         long remoteQuotaTimeMilli = rateLimitInitResponse.getTimestamp();
         long localQuotaTimeMilli = getLocalTimeMilli(remoteQuotaTimeMilli);
         RateLimitWindow rateLimitWindow = initializeRecord.getRateLimitWindow();
+        rateLimitWindow.setLastInitTimeMs(0); // 重置上次初始化时间，从而在metric变更或上报失败时可再次立刻再初始化
         countersList.forEach(counter -> {
             initializeRecord.getDurationRecord().putIfAbsent(counter.getDuration(), counter.getCounterKey());
             counters.putIfAbsent(counter.getCounterKey(),
@@ -238,16 +245,16 @@ private void handleRateLimitInitResponse(RateLimitInitResponse rateLimitInitResp
      *
      * @param rateLimitReportResponse report的回包
      */
-    void handleRateLimitReportResponse(RateLimitReportResponse rateLimitReportResponse) {
+    boolean handleRateLimitReportResponse(RateLimitReportResponse rateLimitReportResponse) {
         LOG.debug("[handleRateLimitReportRequest] response:{}", rateLimitReportResponse);
         if (rateLimitReportResponse.getCode() != ServerCodes.EXECUTE_SUCCESS) {
             LOG.error("[handleRateLimitReportRequest] failed. code is {}", rateLimitReportResponse.getCode());
-            return;
+            return false;
         }
         List<QuotaLeft> quotaLeftsList = rateLimitReportResponse.getQuotaLeftsList();
         if (CollectionUtils.isEmpty(quotaLeftsList)) {
             LOG.error("[handleRateLimitReportRequest] quotaLefts is empty.");
-            return;
+            return true;
         }
         long remoteQuotaTimeMilli = rateLimitReportResponse.getTimestamp();
         long localQuotaTimeMilli = getLocalTimeMilli(remoteQuotaTimeMilli);
@@ -257,6 +264,7 @@ void handleRateLimitReportResponse(RateLimitReportResponse rateLimitReportRespon
                     localQuotaTimeMilli, callback.getDuration() * 1000);
             callback.getRateLimitWindow().getAllocatingBucket().onRemoteUpdate(remoteQuotaInfo);
         });
+        return true;
     }
 
     public int getClientKey() {

polaris-ratelimit/polaris-ratelimit-client/src/main/java/com/tencent/polaris/ratelimit/client/sync/PolarisRemoteSyncTask.java

@@ -57,11 +57,6 @@ public class PolarisRemoteSyncTask implements RemoteSyncTask {
      */
     private final ServiceIdentifier serviceIdentifier;
 
-    /**
-     * 最近一次初始化时间
-     */
-    private long latestInitTime = 0;
-
     public PolarisRemoteSyncTask(RateLimitWindow window) {
         this.window = window;
         this.asyncRateLimitConnector = window.getWindowSet().getAsyncRateLimitConnector();
@@ -83,12 +78,7 @@ public void run() {
                 case DELETED:
                     break;
                 case INITIALIZING:
-                    if (latestInitTime != 0 && System.currentTimeMillis() - latestInitTime < INIT_WAIT_RESPONSE_TIME) {
-                        LOG.debug("currentTime - latestInitTime = {}", System.currentTimeMillis() - latestInitTime);
-                        break;
-                    }
-                    latestInitTime = System.currentTimeMillis();
-                    doRemoteInit();
+                    doRemoteInit(false);
                     break;
                 default:
                     doRemoteAcquire();
@@ -110,7 +100,19 @@ private boolean isInitExpired(InitializeRecord initializeRecord) {
     /**
      * 发送初始化请求
      */
-    private void doRemoteInit() {
+    private void doRemoteInit(boolean redoInit) {
+        // 检查是否在初始化过程中
+        long currentTimeMs = System.currentTimeMillis();
+        long lastInitTime = this.window.getLastInitTimeMs();
+        if (lastInitTime != 0 && currentTimeMs - lastInitTime < INIT_WAIT_RESPONSE_TIME) {
+            LOG.debug("currentTime - latestInitTime = {}", currentTimeMs - lastInitTime);
+            return;
+        }
+        this.window.setLastInitTimeMs(currentTimeMs);
+        if (redoInit) {
+            LOG.warn("[doRemoteAcquire] has not init. redo init: window {}, {}", window, window.getUniqueKey());
+        }
+
         StreamCounterSet streamCounterSet = asyncRateLimitConnector
                 .getStreamCounterSet(window.getWindowSet().getRateLimitExtension().getExtensions(),
                         window.getRemoteCluster(), window.getRemoteAddresses(), window.getUniqueKey(),
@@ -181,8 +183,7 @@ private void doRemoteAcquire() {
         StreamResource streamResource = streamCounterSet.checkAndCreateResource(serviceIdentifier, window);
 
         if (!streamResource.hasInit(serviceIdentifier)) {
-            LOG.warn("[doRemoteAcquire] has not inited. serviceKey:{}", window.getSvcKey());
-            doRemoteInit();
+            doRemoteInit(true);
             return;
         }
         //调整时间
@@ -206,7 +207,7 @@ private void doRemoteAcquire() {
             if (null == counterKey) {
                 LOG.warn("[doRemoteAcquire] counterKey for {}, duration {} not found", window.getUniqueKey(),
                         entry.getKey());
-                doRemoteInit();
+                doRemoteInit(true);
                 return;
             }
             quotaSum.setCounterKey(counterKey);