From 9247578a27af3ba9301126a4fae8f18437f4ff92 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Sun, 6 Sep 2026 23:31:05 +0200 Subject: [PATCH] rest server: serialize access to the shared backend BorgStoreRESTServer is a ThreadingHTTPServer: it handles each connection in its own thread, and every handler operates on the one backend instance shared by all of them (self.server.backend). Each handler does `with self.server.backend:`, i.e. open() on enter and close() on exit, plus one backend operation. With no serialization, concurrent requests to the same server raced: - on the backend's `opened` flag: a second request entering `with backend:` while another still had it open hit `BackendMustNotBeOpen` (HTTP 500), and a close() from one request could tear down the backend under another; - with a quota, on the in-memory usage counter (`_quota_use += delta` is a non-atomic read-modify-write and the limit check reads a value another thread is changing), so usage could drift and the limit be overshot. This only affected the standalone TCP server (e.g. behind nginx) with concurrent connections to the same repository. The stdio-over-ssh server handles requests serially (one process per ssh connection), so it was never affected. Fix: a single per-server lock (self.backend_lock) held around every backend access in the handler - the `with self.server.backend:` operation blocks as well as the create/destroy calls that do not open the backend. This serializes all backend use, mirroring the client-side Store lock. The stdio server gets the lock too (uncontended there, but keeps the handler code uniform). The regression test starts the threaded server with a backend whose store() is slowed (to widen the open()..close() window like real I/O latency) and fires several concurrent requests: without the lock they fail with BackendMustNotBeOpen and the quota drifts; with it all succeed and the quota usage is exact. Co-Authored-By: Claude Opus 4.8 --- src/borgstore/server/rest.py | 37 +++++++++++++-------- tests/test_server_rest.py | 62 +++++++++++++++++++++++++++++++++--- 2 files changed, 81 insertions(+), 18 deletions(-) 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()