Skip to content

rest server: serialize access to the shared backend #220

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ThomasWaldmann merged 1 commit into from
Sep 6, 2026
+81 −18
Merged

rest server: serialize access to the shared backend #220

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 24 additions & 13 deletions src/borgstore/server/rest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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")
Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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")
Expand All @@ -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
)
Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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")
Expand All @@ -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:
Expand All @@ -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)
Expand All @@ -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,
Expand All @@ -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"
Expand Down Expand Up @@ -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)
Expand All @@ -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:
Expand Down
62 changes: 57 additions & 5 deletions tests/test_server_rest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import subprocess
import sys
import threading
import time
import pytest

try:
Expand All @@ -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()

Expand Down Expand Up @@ -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()
Loading