Implement Rendezqueue Server in Python

- Implemented `SwapStore` in `src/rendezqueue/swapstore.py` matching JS logic.
- Implemented `RendezqueueImpl` in `src/rendezqueue/impl.py` for protocol handling.
- Implemented `RendezqueueServer` in `src/rendezqueue/server.py` using `http.server`.
- Added SxPB test cases in `test/server_test_cases.sxpb`.
- Added test runner `test/server_test.py` using `sxpb` library.
- Added integration test `test/integration_test.py` to verify client-server interaction.
- Exposed server classes in `src/rendezqueue/__init__.py`.

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>

Author: gendeux

src/rendezqueue/__init__.py

@@ -1,3 +1,6 @@
 from .client import RendezqueueClient
+from .server import RendezqueueServer, RendezqueueHandler, run as run_server
+from .impl import RendezqueueImpl
+from .swapstore import SwapStore
 
-__all__ = ["RendezqueueClient"]
+__all__ = ["RendezqueueClient", "RendezqueueServer", "RendezqueueHandler", "run_server", "RendezqueueImpl", "SwapStore"]

src/rendezqueue/impl.py

@@ -0,0 +1,172 @@
+import json
+import base64
+import time
+from typing import Optional, Union, Any, Dict
+from rendezqueue.swapstore import SwapStore, TrySwapResponse
+
+MAX_KEY_BYTES = 100
+MAX_ID_BYTES = 100
+MAX_VALUE_BYTES = 65536
+
+def btoa(s: str) -> str:
+    # URL-safe base64 encoding without padding
+    return base64.urlsafe_b64encode(s.encode('latin1')).decode('ascii').rstrip('=')
+
+def atob(s: str) -> str:
+    # URL-safe base64 decoding with padding added
+    missing_padding = len(s) % 4
+    if missing_padding:
+        s += '=' * (4 - missing_padding)
+    return base64.urlsafe_b64decode(s).decode('latin1')
+
+def inplace_decode_tryswap_message(msg: Dict[str, Any]) -> str:
+    if 'b64' not in msg:
+        msg['b64'] = 0
+    elif not isinstance(msg['b64'], int) or msg['b64'] < 0:
+        return "b64"
+
+    if 'ttl' not in msg:
+        msg['ttl'] = 0
+    elif not isinstance(msg['ttl'], int) or msg['ttl'] < 0:
+        return "ttl"
+
+    if 'key' not in msg:
+        msg['key'] = ""
+    elif not isinstance(msg['key'], str):
+        return "key"
+
+    if 'sid' not in msg:
+        msg['sid'] = ""
+    elif not isinstance(msg['sid'], str):
+        return "sid"
+
+    if 'offset' not in msg:
+        msg['offset'] = 0
+    elif not isinstance(msg['offset'], int) or msg['offset'] < 0:
+        return "offset"
+
+    if 'values' not in msg:
+        msg['values'] = []
+    elif not isinstance(msg['values'], list):
+        return "values"
+
+    try:
+        if msg['b64'] & 4:
+            msg['key'] = atob(msg['key'])
+        if msg['b64'] & 2:
+            msg['sid'] = atob(msg['sid'])
+        if msg['b64'] & 1:
+            msg['values'] = [atob(v) for v in msg['values']]
+    except Exception:
+        return "b64_decode_error" # Custom error, JS might just crash or throw
+
+    return ""
+
+def inplace_encode_tryswap_message(msg: Dict[str, Any]) -> None:
+    if 'b64' not in msg:
+        msg['b64'] = 0
+
+    if msg['b64'] & 4:
+        msg['key'] = btoa(msg['key'])
+    if msg['b64'] & 2:
+        msg['sid'] = btoa(msg['sid'])
+    if msg['b64'] & 1:
+        if 'values' not in msg or msg['values'] is None:
+            msg['b64'] &= ~1
+        else:
+            msg['values'] = [btoa(v) for v in msg['values']]
+
+    if msg['b64'] == 0:
+        del msg['b64']
+    if 'ttl' in msg and msg['ttl'] == 0:
+        del msg['ttl']
+    if 'offset' in msg and msg['offset'] == 0:
+        del msg['offset']
+    # If values is None, remove it (TrySwapResponse handles optional)
+    if 'values' in msg and msg['values'] is None:
+        del msg['values']
+
+class RendezqueueImpl:
+    def __init__(self):
+        self.swapstore = SwapStore()
+
+    def tryswap(self, msg: Any, now_ms: Optional[float] = None) -> Union[int, TrySwapResponse]:
+        if not msg:
+            return 400
+
+        # We assume msg is a dict here, validated by inplace_decode... but we should be careful
+        if not isinstance(msg, dict):
+            return 400
+
+        key = msg.get('key', '')
+        sid = msg.get('sid', '')
+        offset = msg.get('offset', 0)
+        values = msg.get('values', [])
+
+        if len(key) > MAX_KEY_BYTES:
+            return 413
+        if len(sid) > MAX_ID_BYTES:
+            return 413
+
+        total_value_bytes = sum(len(v) for v in values)
+        if total_value_bytes > MAX_VALUE_BYTES:
+            return 413
+
+        if now_ms is None:
+            now_ms = time.time() * 1000
+            if now_ms == 0: # unlikely but matches JS
+                return 500
+            now_ms = int(now_ms) # JS uses Math.floor
+
+        result = self.swapstore.tryswap(
+            key, sid, offset, values,
+            now_ms, msg.get('ttl', 0),
+        )
+
+        if not isinstance(result, int):
+            # It's a TrySwapResponse object. We need to convert it to dict for encoding
+            # or wrap it.
+            # But the caller expects to encode it.
+            # Python dataclass to dict:
+            # We can use vars(result) or similar
+            # But we need to preserve b64 from request
+            pass
+
+        return result
+
+    def tryswap_string(self, request_text: str) -> Union[int, str]:
+        msg = None
+        try:
+            msg = json.loads(request_text)
+            e = inplace_decode_tryswap_message(msg)
+            if e:
+                # In JS: throw e
+                # Here we can return error code or message?
+                # The JS code catches 'e' and logs it, then sets msg = null.
+                # So if decoding fails, msg is null.
+                raise Exception(e)
+        except Exception as e:
+            # print(e) # Optional logging
+            msg = None
+
+        result = self.tryswap(msg)
+        if isinstance(result, int):
+            return result
+
+        # Convert dataclass to dict
+        res_dict = {
+            'key': result.key,
+            'sid': result.sid,
+            'offset': result.offset,
+        }
+        if result.values is not None:
+            res_dict['values'] = result.values
+        if result.ttl is not None:
+            res_dict['ttl'] = result.ttl
+
+        # Carry over b64 flags from request if present
+        if msg and 'b64' in msg:
+            res_dict['b64'] = msg['b64']
+
+        inplace_encode_tryswap_message(res_dict)
+        return json.dumps(res_dict)

src/rendezqueue/server.py

@@ -0,0 +1,86 @@
+import http.server
+import socketserver
+import argparse
+from urllib.parse import urlparse
+from rendezqueue.impl import RendezqueueImpl
+
+class RendezqueueHandler(http.server.BaseHTTPRequestHandler):
+    def do_OPTIONS(self):
+        self.send_response(204)
+        if "Access-Control-Request-Headers" in self.headers:
+            self.send_header("Access-Control-Allow-Headers", self.headers["Access-Control-Request-Headers"])
+        if "Access-Control-Request-Method" in self.headers:
+            self.send_header("Access-Control-Allow-Methods", self.headers["Access-Control-Request-Method"])
+        if "Origin" in self.headers:
+            self.send_header("Access-Control-Allow-Origin", self.headers["Origin"])
+        self.end_headers()
+
+    def do_POST(self):
+        parsed_path = urlparse(self.path)
+        if parsed_path.path != self.server.http_path:
+            self.send_error(404, "Not Found")
+            return
+
+        content_type = self.headers.get("Content-Type", "")
+        if content_type != "application/json":
+            self.send_error(418, "I'm a teapot")
+            return
+
+        content_length = int(self.headers.get("Content-Length", 0))
+        body = self.rfile.read(content_length).decode("utf-8")
+
+        result = self.server.impl.tryswap_string(body)
+
+        if isinstance(result, int):
+            self.send_response(result)
+            self.send_header("Access-Control-Allow-Origin", "*")
+            self.send_header("Content-Type", "application/json")
+            self.end_headers()
+        else:
+            self.send_response(200)
+            self.send_header("Access-Control-Allow-Origin", "*")
+            self.send_header("Content-Type", "application/json")
+            self.end_headers()
+            self.wfile.write(result.encode("utf-8"))
+
+class RendezqueueServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
+    def __init__(self, server_address, RequestHandlerClass, impl, http_path="/"):
+        self.impl = impl
+        self.http_path = http_path
+        super().__init__(server_address, RequestHandlerClass)
+
+def run():
+    parser = argparse.ArgumentParser(description="Rendezqueue Server")
+    parser.add_argument("--http_host", default="127.0.0.1", help="Host to bind to")
+    parser.add_argument("--http_port", type=int, default=0, help="Port to bind to (0 for random)")
+    parser.add_argument("--http_path", default="/", help="Path for RPC")
+    parser.add_argument("--port_filepath", help="File to write the port number to")
+
+    # Allow unknown args to be ignored (e.g. for testing or future flags)
+    args, unknown = parser.parse_known_args()
+
+    impl = RendezqueueImpl()
+
+    server = RendezqueueServer(
+        (args.http_host, args.http_port),
+        RendezqueueHandler,
+        impl,
+        http_path=args.http_path
+    )
+
+    host, port = server.server_address
+    print(f"Server running at http://{host}:{port}{args.http_path}")
+
+    if args.port_filepath:
+        with open(args.port_filepath, "w") as f:
+            f.write(str(port))
+
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        server.server_close()
+
+if __name__ == "__main__":
+    run()