forked from
lass.pds.lassul.us/superconfig
This repository has no description
flix.ipfs_upload: fix race between different uploads
author
lassulus
date
(Jun 1, 2026, 9:47 AM CEST)
commit
68130e62
68130e62e57a3b77917b364bbfd5c554085656ec
parent
711a5a24
711a5a2496b9e04005489166663d8a7593c3c14f
+24
+24
2configs/services/flix/ipfs_upload.py
+24
2configs/services/flix/ipfs_upload.py
···
34
34
import socketserver
35
35
import subprocess
36
36
import sys
37
+
import threading
37
38
38
39
DOWNLOAD_ROOT = os.environ.get("DOWNLOAD_ROOT", "/var/lib/ipfs/download")
39
40
STAGING_DIR = os.environ.get("STAGING_DIR", "/var/lib/ipfs/download/incoming/uploader")
···
60
61
return os.path.join(STAGING_DIR, f"{sid}.upload")
61
62
62
63
64
+
_locks_mu = threading.Lock()
65
+
_locks: dict[str, threading.Lock] = {}
66
+
67
+
68
+
def lock_for(rel: str) -> threading.Lock:
69
+
# Serializes concurrent POSTs targeting the same staging file. Without
70
+
# this, two threads handling the same chunk (e.g. a client retry after a
71
+
# lost response) can both pass the start==current_size check and both
72
+
# open("ab"), duplicating bytes and shifting the resume offset.
73
+
with _locks_mu:
74
+
lock = _locks.get(rel)
75
+
if lock is None:
76
+
lock = threading.Lock()
77
+
_locks[rel] = lock
78
+
return lock
79
+
80
+
63
81
class Handler(http.server.BaseHTTPRequestHandler):
64
82
server_version = "ipfs-upload/1"
65
83
···
81
99
cr_header = self.headers.get("Content-Range")
82
100
staging = staging_path(rel)
83
101
102
+
with lock_for(rel):
103
+
self._handle_locked(rel, length, cr_header, staging)
104
+
105
+
def _handle_locked(
106
+
self, rel: str, length: int, cr_header: str | None, staging: str
107
+
) -> None:
84
108
# Status check: empty body + no Content-Range.
85
109
if length == 0 and not cr_header:
86
110
try: