diff --git a/src/borgstore/server/rest.py b/src/borgstore/server/rest.py index 1b456b9..de5327e 100644 --- a/src/borgstore/server/rest.py +++ b/src/borgstore/server/rest.py @@ -8,6 +8,7 @@ import socket import sys import itertools +import threading from http import HTTPStatus as HTTP from http.server import ThreadingHTTPServer, HTTPServer, BaseHTTPRequestHandler from pathlib import Path @@ -171,7 +172,8 @@ def do_POST(self): cmd = self.query.get("cmd", [None])[0] if cmd == "create": try: - self.server.backend.create() + with self.server.backend_lock: + self.server.backend.create() self.respond(HTTP.OK) except Exception as e: self._handle_exception(e, "create") @@ -182,7 +184,7 @@ def do_POST(self): new = self.query.get("new", [None])[0] if current and new: try: - with self.server.backend: + with self.server.backend_lock, self.server.backend: self.server.backend.move(current, new) self.respond(HTTP.OK) except Exception as e: @@ -193,7 +195,7 @@ def do_POST(self): if cmd == "mkdir": try: - with self.server.backend: + with self.server.backend_lock, self.server.backend: self.server.backend.mkdir(self.name) self.respond(HTTP.OK) except Exception as e: @@ -206,7 +208,7 @@ def do_POST(self): return algorithm = self.query.get("algorithm", ["sha256"])[0] try: - with self.server.backend: + with self.server.backend_lock, self.server.backend: digest = self.server.backend.hash(self.name, algorithm=algorithm) self.respond(HTTP.OK, data=digest.encode("ascii"), content_type="text/plain") except Exception as e: @@ -215,7 +217,7 @@ def do_POST(self): if cmd == "quota": try: - with self.server.backend: + with self.server.backend_lock, self.server.backend: quota_info = self.server.backend.quota() response_data = json.dumps(quota_info).encode("utf-8") self.respond(HTTP.OK, data=response_data, content_type="application/json") @@ -235,7 +237,7 @@ def do_POST(self): content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) sources = json.loads(body) - with self.server.backend: + with self.server.backend_lock, self.server.backend: target = self.server.backend.defrag( sources, target=target, algorithm=algorithm, namespace=namespace, levels=levels ) @@ -257,7 +259,7 @@ def do_POST(self): if got_hash != expected_hash: self.respond(HTTP.UNPROCESSABLE_ENTITY, b"Content hash verification failed, please retry") return - with self.server.backend: + with self.server.backend_lock, self.server.backend: self.server.backend.store(self.name, data) self.respond(HTTP.OK) except Exception as e: @@ -271,7 +273,7 @@ def do_DELETE(self): cmd = self.query.get("cmd", [None])[0] if cmd == "rmdir": try: - with self.server.backend: + with self.server.backend_lock, self.server.backend: self.server.backend.rmdir(self.name) self.respond(HTTP.OK) except Exception as e: @@ -280,7 +282,8 @@ def do_DELETE(self): if cmd == "destroy": try: - self.server.backend.destroy() + with self.server.backend_lock: + self.server.backend.destroy() self.respond(HTTP.OK) except Exception as e: self._handle_exception(e, "destroy") @@ -291,7 +294,7 @@ def do_DELETE(self): return try: - with self.server.backend: + with self.server.backend_lock, self.server.backend: self.server.backend.delete(self.name) self.respond(HTTP.OK) except Exception as e: @@ -300,7 +303,7 @@ def do_DELETE(self): @checks_and_logging def do_HEAD(self): try: - with self.server.backend: + with self.server.backend_lock, self.server.backend: info = self.server.backend.info(self.name) if not info.exists: raise ObjectNotFound(self.name) @@ -323,7 +326,7 @@ def do_GET(self): try: # send a JSON list of objects # [{"name": "...", "size": ...}, ...] - with self.server.backend: + with self.server.backend_lock, self.server.backend: items = ( { "name": item.name, @@ -350,7 +353,7 @@ def do_GET(self): range_header = self.headers.get("Range") offset, size = parse_range_header(range_header) if range_header else (0, None) - with self.server.backend: + with self.server.backend_lock, self.server.backend: data = self.server.backend.load(self.name, offset=offset, size=size) self.respond( HTTP.PARTIAL_CONTENT if range_header else HTTP.OK, data=data, content_type="application/octet-stream" @@ -492,6 +495,10 @@ def serve_forever(self, poll_interval=0.5): class BorgStoreStdioRESTServer(StdIOHTTPServer): def __init__(self, backend, username=None, password=None): self.backend = backend + # serialize all access to the single shared backend instance: the server is threaded + # (a thread per request) and the backend is neither thread-safe nor safe to open/close + # concurrently, so every handler holds this lock around its `with self.backend:` block. + self.backend_lock = threading.Lock() self.username = username self.password = password super().__init__(BorgStoreRESTRequestHandler) @@ -510,6 +517,10 @@ class BorgStoreRESTServer(ThreadingHTTPServer): def __init__(self, server_address, backend, username=None, password=None, adopted_socket=None): self.backend = backend + # serialize all access to the single shared backend instance: the server is threaded + # (a thread per request) and the backend is neither thread-safe nor safe to open/close + # concurrently, so every handler holds this lock around its `with self.backend:` block. + self.backend_lock = threading.Lock() self.username = username self.password = password if adopted_socket is not None: diff --git a/tests/test_server_rest.py b/tests/test_server_rest.py index 5bef820..0ecfcb1 100644 --- a/tests/test_server_rest.py +++ b/tests/test_server_rest.py @@ -3,6 +3,7 @@ import subprocess import sys import threading +import time import pytest try: @@ -17,18 +18,19 @@ blake3_is_available = blake3 is not None -from borgstore.constants import DEL_SUFFIX +from borgstore.constants import DEL_SUFFIX, QUOTA_STORE_NAME from borgstore.server.rest import BorgStoreRESTServer from borgstore.backends.rest import get_rest_backend -from borgstore.backends.posixfs import get_file_backend +from borgstore.backends.posixfs import get_file_backend, PosixFS from borgstore.backends.errors import ObjectNotFound, BackendAlreadyExists, QuotaExceeded, ReadRangeError from borgstore.store import get_backend, Store -def start_server(backend_url, address, port, username=None, password=None, permissions=None, quota=None): - from borgstore.store import get_backend +def start_server(backend_url, address, port, username=None, password=None, permissions=None, quota=None, backend=None): + if backend is None: + from borgstore.store import get_backend - backend = get_backend(backend_url, permissions=permissions, quota=quota) + backend = get_backend(backend_url, permissions=permissions, quota=quota) server = BorgStoreRESTServer((address, port), backend, username, password) ready = threading.Event() @@ -763,3 +765,53 @@ def test_rest_url(tmp_path): assert info_root.directory store.destroy() + + +def test_concurrent_requests_share_one_backend_safely(tmp_path): + # The threaded REST server hands one shared backend instance to all request threads. Each request + # does `with backend:` (open on enter, close on exit) plus one operation, so without serialization + # concurrent requests collide on the backend's `opened` flag (BackendMustNotBeOpen) and, with a + # quota, on the in-memory usage counter. A slow store widens that window so the collision is + # reliable; with the server-side backend lock all requests succeed and the quota usage is exact. + store_path = tmp_path / "store" + + class SlowStorePosixFS(PosixFS): + def store(self, name, value): + time.sleep(0.05) # widen the open()..close() window, like real I/O latency + return super().store(name, value) + + n, size = 6, 100 + backend = SlowStorePosixFS(store_path, quota=10**9) + backend.create() + + server, thread = start_server(None, "127.0.0.1", 0, backend=backend) + host, port = server.server_address + url = f"http://{host}:{port}/" + errors = [] + barrier = threading.Barrier(n) + + def worker(i): + try: + be = get_rest_backend(url) + be.open() + try: + barrier.wait(timeout=10) # release all requests as simultaneously as possible + be.store(f"key{i}", b"x" * size) + finally: + be.close() + except Exception as e: + errors.append(repr(e)) + + try: + threads = [threading.Thread(target=worker, args=(i,)) for i in range(n)] + for th in threads: + th.start() + for th in threads: + th.join(timeout=30) + assert not any(th.is_alive() for th in threads) + assert errors == [], f"concurrent requests failed: {errors}" + # every store landed and the quota usage is exact (no failed or lost updates): + assert int((store_path / QUOTA_STORE_NAME).read_text()) == n * size + finally: + server.shutdown() + server.server_close()