chore: 优化 scheduler 的原因，导致没有预期上线

Author: tianxiu2b2t

assets/js/index.js

@@ -1213,6 +1213,11 @@ class UserAuth {
             // ...
         })
     }
+    get triggerGUI() {
+        return SVGContainers.user.addEventListener("click", () => {
+            // ...
+        })
+    }
 }
 const $userAuth = new UserAuth();
 
@@ -1233,7 +1238,8 @@ async function load() {
             createElement("h2").text(document.title)
         ),
         createElement("div").classes("content").append(
-            $theme_change
+            $theme_change,
+            $userAuth.triggerGUI,
         )
     )
     const $app = createElement("div").classes("app")

core/__init__.py

@@ -3,11 +3,11 @@
 import time
 
 from core import web
+from . import scheduler
 from .logger import logger
 from . import utils
 import atexit
 from . import cluster
-from . import scheduler
 from . import dashboard
 from . import database
 from . import storages

core/cluster.py

@@ -854,7 +854,7 @@ async def enable(self):
         if not self.initialized:
             return
         for cluster in self.clusters:
-            if cluster.delay_enable_task is not None:
+            if cluster.delay_enable_task is not None or cluster.delay >= time.monotonic():
                 continue
             await cluster.enable()
 
@@ -944,6 +944,7 @@ def __init__(self, id: str, secret: str):
         self.enabled = False
         self.enable_count: int = 0
         self.keepalive_error: int = 0
+        self.delay = 0
         self.keepalive_task: Optional[int] = None
         self.delay_enable_task: Optional[int] = None
         self.counter: defaultdict[storages.iStorage, ClusterCounter] = defaultdict(ClusterCounter)
@@ -1121,6 +1122,7 @@ async def disable(self, exit: bool = False):
 
     def retry(self):
         delay = min(3600, ((self.enable_count  + 1) ** 2) * 60)
+        self.delay = delay + time.monotonic()
         self.delay_enable_task = scheduler.run_later(self.enable, delay)
         logger.tinfo("cluster.info.cluster.retry_enable", cluster=self.id, delay=units.format_count_datetime(delay))
 

core/dashboard.py

@@ -183,6 +183,36 @@ def __init__(
             300
         )
         self._loop = loop
+    
+    def start_keepalive(self):
+        return
+        scheduler.run_repeat_later(
+            self.keepalive,
+            2,
+            30
+        )
+
+    async def send_keepalive(
+        self,
+        client: SSEClient
+    ):
+        current = str(utils.ObjectId())
+        await self.send_message_client(
+            client,
+            current,
+            "keepalive",
+            current
+        )
+        
+
+    async def keepalive(
+        self
+    ):
+        for client in self._connections.copy():
+            try:
+                await self.send_keepalive(client)
+            except:
+                await self.close_client(client)
 
     async def send(
         self,
@@ -195,6 +225,17 @@ async def send(
             )
 
     async def raw_send_client(
+        self,
+        client: SSEClient,
+        data: str
+    ):
+        if len(data) == 0:
+            return
+        for _ in range(2 - data[-2:].count("\n")):
+            data += "\n"
+        await client.resp.write(data.encode("utf-8"))
+
+    async def send_message_client(
         self,
         client: SSEClient,
         id: Optional[str] = None,
@@ -209,8 +250,10 @@ async def raw_send_client(
         }.items():
             if item[1] is not None:
                 buffer.write(f"{item[0]}: {item[1]}\n")
-        buffer.write("\n")
-        await client.resp.write(buffer.getvalue().encode("utf-8"))
+        await self.raw_send_client(
+            client,
+            buffer.getvalue()
+        )
 
     async def send_client(
         self,
@@ -224,7 +267,7 @@ async def send_client(
             await self.close_client(client)
             return
         try:
-            await self.raw_send_client(
+            await self.send_message_client(
                 client,
                 id=str(message.id),
                 event="message",
@@ -236,9 +279,9 @@ async def send_client(
                 )
             )
         except:
+            logger.traceback()
             await self.close_client(client)
 
-
     async def request(
         self,
         request: web.Request,
@@ -260,13 +303,22 @@ async def request(
             await resp.prepare(request)
             self._connections.append(client)
             last_event_id: str = request.headers.get("Last-Event-ID", "0" * 24)
-            last_id = utils.ObjectId(last_event_id)
+            try:
+                last_id = utils.ObjectId(last_event_id)
+            except:
+                last_id = utils.ObjectId("0" * 24)
+            
+            #await self.send_keepalive(client)
+
             for message in self._cache.keys():
                 if message.generation_time <= last_id.generation_time:
                     break
+                msg = self._cache.get(message)
+                if msg is cache.EMPTY:
+                    continue
                 await self.send_client(
                     client,
-                    self._cache.get(message)
+                    msg
                 )
 
             try:
@@ -964,26 +1016,46 @@ def query_geo(
     cn: bool = False
 ):
     resp_data: defaultdict[str, int] = defaultdict(int)
+    since_hour = get_query_hour_tohour(day * 24)
+    with db.SESSION as session:
+        q = session.query(db.ResponseTable).filter(
+            db.ResponseTable.hour >= since_hour
+        ).order_by(db.ResponseTable.hour)
+        for item in q.all():
+            wait_threads(
+                thread_decompress_data_and_query,
+                [
+                    (item.ip_tables, cn, resp_data)
+                ]
+            )
+    
+    return resp_data
+
+
+def wait_threads(
+    target,
+    args: list[tuple[Any, ...]] = []
+):
     threads: list[threading.Thread] = []
-    for hour, data in query_addresses(day * 24).items():
+    for arg in args:
         threads.append(threading.Thread(
-            target=thread_query_geo,
-            args=(data, resp_data, cn)
+            target=target,
+            args=arg
         ))
     for thread in threads:
         thread.start()
     for thread in threads:
         thread.join()
     threads.clear()
-    return resp_data
 
-def thread_query_geo(
-    data: dict[str, int],
-    resp_data: defaultdict[str, int],
-    cn: bool
+def thread_decompress_data_and_query(
+    origin: bytes,
+    cn,
+    resp_data: defaultdict[str, int]
 ):
-    for ip, count in data.items():
-        address = query_ip(ip)
+    data = db.decompress(origin)
+    for k, count in data.items():
+        address = query_ip(k)
         if cn and address.country == "CN":
             resp_data[address.province] += count
         else:
@@ -1068,6 +1140,7 @@ def record():
 
 async def init():
     global task, status_task
+    SSEEMIT.start_keepalive()
     task = threading.Thread(target=record)
     task.start()
     status_task = asyncio.create_task(push_status())