implement features needed for caching in ranking

cmscommon/eventsource.py

@@ -96,6 +96,8 @@ def __init__(self, size: int):
         # and have the ones at the other end be dropped when the total
         # number exceeds the given limit.
         self._cache = deque(maxlen=size)
+        key = int(time.time() * 1_000_000)
+        self._cache.append((key, None))
         # We use a WeakSet as we want queues to be vanish automatically
         # when no one else is using (i.e. fetching from) them.
         self._sub_queues = WeakSet()
@@ -141,11 +143,13 @@ def get_subscriber(self, last_event_id: str | None = None) -> "Subscriber":
             if len(self._cache) > 0 and last_event_key >= self._cache[0][0]:
                 # All missed events are in cache.
                 for key, msg in self._cache:
+                    if msg is None:
+                        continue
                     if key > last_event_key:
                         queue.put(msg)
             else:
                 # Some events may be missing. Ask to reinit.
-                queue.put(b"event:reinit\n\n")
+                queue.put(b"event:reinit\ndata:reinit\n\n")
         # Store the queue and return a subscriber bound to it.
         self._sub_queues.add(queue)
         return Subscriber(queue)
@@ -211,14 +215,17 @@ class EventSource:
     _GLOBAL_TIMEOUT = 600
     _WRITE_TIMEOUT = 30
     _PING_TIMEOUT = 15
+    _ONE_SHOT_POLL_TIMEOUT = 5
 
     _CACHE_SIZE = 250
 
-    def __init__(self):
+    def __init__(self, retry_backoff_seconds: int = 0, cache_interval: int = 1):
         """Create an event source.
 
         """
         self._pub = Publisher(self._CACHE_SIZE)
+        self.retry_backoff_seconds = retry_backoff_seconds
+        self.cache_interval = cache_interval
 
     def send(self, event: str, data: str):
         """Send the event to the stream.
@@ -298,22 +305,6 @@ def wsgi_app(self, environ, start_response):
         if request.accept_mimetypes.quality("text/event-stream") <= 0:
             return NotAcceptable()(environ, start_response)
 
-        # Initialize the response and get the write() callback. The
-        # Cache-Control header is useless for conforming clients, as
-        # the spec. already imposes that behavior on them, but we set
-        # it explicitly to avoid unwanted caching by unaware proxies and
-        # middlewares.
-        write = start_response(
-            "200 OK", [("Content-Type", "text/event-stream; charset=utf-8"),
-                       ("Cache-Control", "no-cache")])
-
-        # This is a part of the fourth hack (see above).
-        if hasattr(start_response, "__self__") and \
-                isinstance(start_response.__self__, WSGIHandler):
-            handler = start_response.__self__
-        else:
-            handler = None
-
         # One-shot means that we will terminate the request after the
         # first batch of sent events. We do this when we believe the
         # client doesn't support chunked transfer. As this encoding has
@@ -327,11 +318,35 @@ def wsgi_app(self, environ, start_response):
         # newer werkzeug version. But all modern browsers support SSE natively
         # so this check isn't necessary nowadays. (Well, the http/1.1 check
         # probably isn't necessary either, to be honest...)
-        if environ["SERVER_PROTOCOL"] != "HTTP/1.1":
+        if (environ["SERVER_PROTOCOL"] != "HTTP/1.1" or
+            request.headers.get("X-Response-Mode", "eventstream") == "one-shot"):
             one_shot = True
         else:
             one_shot = False
 
+        if one_shot:
+            ping_timeout = self._ONE_SHOT_POLL_TIMEOUT
+            # Allow CDNs to reuse finished one-shot responses briefly.
+            # Empty responses can stick for at most this long.
+            max_age = max(0, self.cache_interval - 1)
+            cache_control = "public, max-age=%d" % max_age
+        else:
+            ping_timeout = self._PING_TIMEOUT
+            # Long-lived streams must not be cached by intermediaries.
+            cache_control = "no-cache"
+
+        # Initialize the response and get the write() callback.
+        write = start_response(
+            "200 OK", [("Content-Type", "text/event-stream; charset=utf-8"),
+                       ("Cache-Control", cache_control)])
+
+        # This is a part of the fourth hack (see above).
+        if hasattr(start_response, "__self__") and \
+                isinstance(start_response.__self__, WSGIHandler):
+            handler = start_response.__self__
+        else:
+            handler = None
+
         # As for the Server-Sent Events [1] spec., this is the way for
         # the client to tell us the ID of the last event it received
         # and to ask us to send it the ones that happened since then.
@@ -348,6 +363,12 @@ def wsgi_app(self, environ, start_response):
 
         # We subscribe to the publisher to receive events.
         sub = self._pub.get_subscriber(last_event_id)
+
+
+        if self.retry_backoff_seconds > 0:
+            # Sned how long the client should wait before retrying
+            # the connection.
+            write("retry: {}\n".format(self.retry_backoff_seconds * 1000).encode("utf-8"))
 
         # Send some data down the pipe. We need that to make the user
         # agent announces the connection (see the spec.). Since it's a
@@ -369,7 +390,7 @@ def wsgi_app(self, environ, start_response):
                 # seconds, sending a ping (i.e. a comment) if there's
                 # no real data.
                 try:
-                    with Timeout(self._PING_TIMEOUT):
+                    with Timeout(ping_timeout):
                         data = b"".join(sub.get())
                         got_sth = True
                 except Timeout:
@@ -391,10 +412,8 @@ def wsgi_app(self, environ, start_response):
                         handler.response_use_chunked = False
                     break
 
-                # If we decided this is one-shot, stop the long-poll as
-                # soon as we sent the client some real data.
-                if one_shot and got_sth:
+                # If we decided this is one-shot, stop the long-poll.
+                if one_shot:
                     break
-
         # An empty iterable tells the server not to send anything.
         return []

cmsranking/Config.py

@@ -40,6 +40,10 @@ def default_path(name):
 @dataclass
 class PublicConfig:
     show_id_column: bool = False
+    flags_extension: str = ""
+    faces_extension: str = ""
+    cache_interval: int = 1
+    retry_backoff_seconds: int = 0
 
 
 @dataclass

cmsranking/RankingWebServer.py

@@ -59,6 +59,11 @@
 logger = logging.getLogger(__name__)
 
 
+def public_cache_control(cache_interval: int) -> str:
+    """Build a Cache-Control value with max-age of cache_interval - 1."""
+    return "public, max-age=%d" % max(0, cache_interval - 1)
+
+
 class CustomUnauthorized(Unauthorized):
 
     def __init__(self, realm_name: str):
@@ -73,11 +78,19 @@ def get_response(self, environ=None):
 
 class StoreHandler:
 
-    def __init__(self, store: Store, username: str, password: str, realm_name: str):
+    def __init__(
+        self,
+        store: Store,
+        username: str,
+        password: str,
+        realm_name: str,
+        cache_interval: int = 1,
+    ):
         self.store = store
         self.username = username
         self.password = password
         self.realm_name = realm_name
+        self.cache_interval = cache_interval
 
         self.router = Map([
             Rule("/<key>", methods=["GET"], endpoint="get"),
@@ -138,12 +151,14 @@ def get(self, request: Request, response: Response, key: str):
 
         response.status_code = 200
         response.headers['Timestamp'] = "%0.6f" % time.time()
+        response.headers['Cache-Control'] = public_cache_control(self.cache_interval)
         response.mimetype = "application/json"
         response.data = json.dumps(self.store.retrieve(key))
 
     def get_list(self, request: Request, response: Response):
         response.status_code = 200
         response.headers['Timestamp'] = "%0.6f" % time.time()
+        response.headers['Cache-Control'] = public_cache_control(self.cache_interval)
         response.mimetype = "application/json"
         response.data = json.dumps(self.store.retrieve_list())
 
@@ -242,9 +257,15 @@ def delete_list(self, request: Request, response: Response):
 class DataWatcher(EventSource):
     """Receive the messages from the entities store and redirect them."""
 
-    def __init__(self, stores: dict[str, Store], buffer_size: int):
+    def __init__(
+        self,
+        stores: dict[str, Store],
+        buffer_size: int,
+        retry_backoff_seconds: int = 0,
+        cache_interval: int = 1,
+    ):
         self._CACHE_SIZE = buffer_size
-        EventSource.__init__(self)
+        EventSource.__init__(self, retry_backoff_seconds, cache_interval)
 
         stores["contest"].add_create_callback(
             functools.partial(self.callback, "contest", "create"))
@@ -285,9 +306,10 @@ def score_callback(self, user: str, task: str, score: float):
 
 class SubListHandler:
 
-    def __init__(self, stores: dict[str, Store]):
+    def __init__(self, stores: dict[str, Store], cache_interval: int = 1):
         self.task_store: Store[Task] = stores["task"]
         self.scoring_store: ScoringStore = stores["scoring"]
+        self.cache_interval = cache_interval
 
         self.router = Map([
             Rule("/<user_id>", methods=["GET"], endpoint="sublist"),
@@ -322,6 +344,7 @@ def wsgi_app(self, environ, start_response):
 
         response = Response()
         response.status_code = 200
+        response.headers['Cache-Control'] = public_cache_control(self.cache_interval)
         response.mimetype = "application/json"
         response.data = json.dumps(result)
 
@@ -330,8 +353,9 @@ def wsgi_app(self, environ, start_response):
 
 class HistoryHandler:
 
-    def __init__(self, stores: dict[str, Store]):
+    def __init__(self, stores: dict[str, Store], cache_interval: int = 1):
         self.scoring_store: ScoringStore = stores["scoring"]
+        self.cache_interval = cache_interval
 
     def __call__(self, environ, start_response):
         return self.wsgi_app(environ, start_response)
@@ -346,6 +370,7 @@ def wsgi_app(self, environ, start_response):
 
         response = Response()
         response.status_code = 200
+        response.headers['Cache-Control'] = public_cache_control(self.cache_interval)
         response.mimetype = "application/json"
         response.data = json.dumps(result)
 
@@ -354,8 +379,9 @@ def wsgi_app(self, environ, start_response):
 
 class ScoreHandler:
 
-    def __init__(self, stores: dict[str, Store]):
+    def __init__(self, stores: dict[str, Store], cache_interval: int = 1):
         self.scoring_store: ScoringStore = stores["scoring"]
+        self.cache_interval = cache_interval
 
     def __call__(self, environ, start_response):
         return self.wsgi_app(environ, start_response)
@@ -375,6 +401,7 @@ def wsgi_app(self, environ, start_response):
         response = Response()
         response.status_code = 200
         response.headers['Timestamp'] = "%0.6f" % time.time()
+        response.headers['Cache-Control'] = public_cache_control(self.cache_interval)
         response.mimetype = "application/json"
         response.data = json.dumps(result)
 
@@ -411,24 +438,39 @@ def wsgi_app(self, environ, start_response):
             return exc
 
         location = self.location % args
-
         request = Request(environ)
+        request.encoding_errors = "strict"
+
+        # Determine which path to serve
+        path = None
+
+        if os.path.isfile(location):
+            # Exact path exists, check if extension is supported
+            _, ext = os.path.splitext(location)
+            ext = ext.lstrip('.')
+            if ext in self.EXT_TO_MIME:
+                path = location
+                mimetype = self.EXT_TO_MIME[ext]
+
+        if path is None:
+            # Check available extensions
+            available: list[str] = list()
+            for extension, mimetype in self.EXT_TO_MIME.items():
+                if os.path.isfile(location + '.' + extension):
+                    available.append(mimetype)
+
+            mimetype = request.accept_mimetypes.best_match(available)
+            if mimetype is not None:
+                path = "%s.%s" % (location, self.MIME_TO_EXT[mimetype])
+            else:
+                path = self.fallback
+                mimetype = 'image/png'  # FIXME Hardcoded type.
 
+        # Serve the file
         response = Response()
-
-        available: list[str] = list()
-        for extension, mimetype in self.EXT_TO_MIME.items():
-            if os.path.isfile(location + '.' + extension):
-                available.append(mimetype)
-        mimetype = request.accept_mimetypes.best_match(available)
-        if mimetype is not None:
-            path = "%s.%s" % (location, self.MIME_TO_EXT[mimetype])
-        else:
-            path = self.fallback
-            mimetype = 'image/png'  # FIXME Hardcoded type.
-
         response.status_code = 200
         response.mimetype = mimetype
+
         response.last_modified = \
             datetime.utcfromtimestamp(os.path.getmtime(path))\
                     .replace(microsecond=0)
@@ -480,6 +522,8 @@ def wsgi_app(self, environ, start_response):
 
         response = Response()
         response.status_code = 200
+        response.headers['Cache-Control'] = public_cache_control(
+            self.pub_config.cache_interval)
         response.mimetype = "application/json"
         print(str(self.pub_config))
         response.data = json.dumps(
@@ -609,43 +653,53 @@ def main() -> int:
     stores["scoring"] = ScoringStore(stores)
     stores["scoring"].init_store()
 
+    cache_interval = config.public.cache_interval
+
     toplevel_handler = RoutingHandler(
         RootHandler(web_dir),
-        DataWatcher(stores, config.buffer_size),
+        DataWatcher(
+            stores, config.buffer_size, config.public.retry_backoff_seconds,
+            cache_interval),
         ImageHandler(
             os.path.join(config.lib_dir, '%(name)s'),
             os.path.join(web_dir, 'img', 'logo.png')),
-        ScoreHandler(stores),
-        HistoryHandler(stores),
+        ScoreHandler(stores, cache_interval),
+        HistoryHandler(stores, cache_interval),
         PublicConfigHandler(config.public))
 
     wsgi_app = SharedDataMiddleware(DispatcherMiddleware(
         toplevel_handler, {
             '/contests': StoreHandler(
                 stores["contest"],
-                config.username, config.password, config.realm_name),
+                config.username, config.password, config.realm_name,
+                cache_interval),
             '/tasks': StoreHandler(
                 stores["task"],
-                config.username, config.password, config.realm_name),
+                config.username, config.password, config.realm_name,
+                cache_interval),
             '/teams': StoreHandler(
                 stores["team"],
-                config.username, config.password, config.realm_name),
+                config.username, config.password, config.realm_name,
+                cache_interval),
             '/users': StoreHandler(
                 stores["user"],
-                config.username, config.password, config.realm_name),
+                config.username, config.password, config.realm_name,
+                cache_interval),
             '/submissions': StoreHandler(
                 stores["submission"],
-                config.username, config.password, config.realm_name),
+                config.username, config.password, config.realm_name,
+                cache_interval),
             '/subchanges': StoreHandler(
                 stores["subchange"],
-                config.username, config.password, config.realm_name),
+                config.username, config.password, config.realm_name,
+                cache_interval),
             '/faces': ImageHandler(
                 os.path.join(config.lib_dir, 'faces', '%(name)s'),
                 os.path.join(web_dir, 'img', 'face.png')),
             '/flags': ImageHandler(
                 os.path.join(config.lib_dir, 'flags', '%(name)s'),
                 os.path.join(web_dir, 'img', 'flag.png')),
-            '/sublist': SubListHandler(stores),
+            '/sublist': SubListHandler(stores, cache_interval),
         }), {'/': web_dir})
 
     servers: list[WSGIServer] = list()