27470493850

Committed 13 Jun 2026 03:09PM UTC coverage: 94.778% (+0.2%) from 94.575%

Build # 27470493850

Build Type

push

github

Committed by

web-flow

Commit Message

feat: Add upload lock contention resolution for HEAD requests (#80)

* feat: Add lock contention resolution for HEAD requests and update AGENTS.md guidelines

* docs: Simplify CHANGELOG.md entry and clean up structure of AGENTS.md

* docs: Add TESTING.md and format locking/agents docs

* docs: Update TESTING.md with build, run, and cleanup instructions

* docs: Add dependency update step in TESTING.md

* feat: Reviewing

* test: Add unit tests to cover missed lines in DiskLockingService, InterruptibleInputStream, UploadLockingService, and TusFileUploadService

* Add maven profile and python script to check unit test coverage locally

* Support brief range grouping and git branch comparison in coverage check

* Add instruction in AGENTS.md to check coverage on new lines

* Add extra unit tests to achieve 100% line coverage on modified files

* Support multiple Jacoco XML report inputs in coverage check profile

Coverage Stats

614 of 692 branches covered (88.73%)

Branch coverage included in aggregate %.

198 of 200 new or added lines in 7 files covered. (99.0%)

1800 of 1855 relevant lines covered (97.04%)

6.57 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.3

/src/main/java/me/desair/tus/server/upload/disk/DiskLockingService.java

package me.desair.tus.server.upload.disk;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.lang.ref.WeakReference;
import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.attribute.FileTime;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import me.desair.tus.server.exception.TusException;
import me.desair.tus.server.exception.UploadAlreadyLockedException;
import me.desair.tus.server.upload.UploadId;
import me.desair.tus.server.upload.UploadIdFactory;
import me.desair.tus.server.upload.UploadLock;
import me.desair.tus.server.upload.UploadLockingService;
import me.desair.tus.server.util.InterruptibleInputStream;
import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * {@link UploadLockingService} implementation that uses the file system for implementing locking
 * <br>
 * File locking can also apply to shared network drives. This way the framework supports clustering
 * as long as the upload storage directory is mounted as a shared (network) drive. <br>
 * File locks are also automatically released on application (JVM) shutdown. This means the file
 * locking is not persistent and prevents cleanup and stale lock issues.
 */
public class DiskLockingService extends AbstractDiskBasedService implements UploadLockingService {

  private static final Logger log = LoggerFactory.getLogger(DiskLockingService.class);
  private static final String LOCK_SUB_DIRECTORY = "locks";

  /** Registry tracking active request input streams in the current JVM. */
  private static final ConcurrentHashMap<String, WeakReference<InterruptibleInputStream>>
      activeLocks = new ConcurrentHashMap<>();

  private static Thread watchdogThread = null;
  private static final Object watchdogLock = new Object();

  private UploadIdFactory idFactory;

  public DiskLockingService(String storagePath) {
    super(storagePath + File.separator + LOCK_SUB_DIRECTORY);
  }

  /** Constructor to use custom UploadIdFactory. */
  public DiskLockingService(UploadIdFactory idFactory, String storagePath) {
    this(storagePath);
    Validate.notNull(idFactory, "The IdFactory cannot be null");
    this.idFactory = idFactory;
  }

  /**
   * Attempts to lock the upload resource. Wraps the lock in a RegisteredLock decorator to manage
   * cleanup of stop files and the active lock registry.
   */
  @Override
  public UploadLock lockUploadByUri(String requestUri) throws TusException, IOException {

    UploadId id = idFactory.readUploadId(requestUri);

    UploadLock lock = null;

    Path lockPath = getLockPath(id);
    // If lockPath is not null, we know this is a valid Upload URI
    if (lockPath != null) {
      FileBasedLock baseLock = new FileBasedLock(requestUri, lockPath);
      Path stopFilePath = baseLock.getLockPath().resolveSibling(id.toString() + ".stop");
      lock = new RegisteredLock(baseLock, id.toString(), stopFilePath);
    }
    return lock;
  }

  /** Cleans up stale locks and stop files in the storage directory. */
  @Override
  public void cleanupStaleLocks() throws IOException {
    try (DirectoryStream<Path> locksStream = Files.newDirectoryStream(getStoragePath())) {
      for (Path path : locksStream) {
        if (Files.exists(path)) {
          FileTime lastModifiedTime = Files.getLastModifiedTime(path);
          if (lastModifiedTime.toMillis() < System.currentTimeMillis() - 10000L) {
            String fileName = path.getFileName().toString();
            if (fileName.endsWith(".stop")) {
              Files.deleteIfExists(path);
            } else {
              UploadId id = new UploadId(fileName);
              if (!isLocked(id)) {
                Files.deleteIfExists(path);
                Files.deleteIfExists(path.resolveSibling(fileName + ".stop"));
              }
            }
          }
        }
      }
    }
  }

  /** Checks whether the upload is locked by attempting to obtain a short-lived file lock. */
  @Override
  public boolean isLocked(UploadId id) {
    boolean locked = false;
    Path lockPath = getLockPath(id);

    if (lockPath != null) {
      // Try to obtain a lock to see if the upload is currently locked
      try (UploadLock lock = new FileBasedLock(id.toString(), lockPath)) {

        // We got the lock, so it means no one else is locking it.
        locked = false;

      } catch (UploadAlreadyLockedException | IOException e) {
        // There was already a lock
        locked = true;
      }
    }

    return locked;
  }

  @Override
  public void setIdFactory(UploadIdFactory idFactory) {
    Validate.notNull(idFactory, "The IdFactory cannot be null");
    this.idFactory = idFactory;
  }

  /**
   * Registers an active request input stream so that it can be interrupted if lock contention
   * occurs.
   */
  @Override
  public void registerInputStream(String requestUri, InputStream inputStream) {
    if (inputStream == null) {
      return;
    }
    UploadId id = idFactory.readUploadId(requestUri);
    if (id == null) {
      return;
    }
    if (inputStream instanceof InterruptibleInputStream) {
      activeLocks.put(id.toString(), new WeakReference<>((InterruptibleInputStream) inputStream));
      startWatchdogIfNecessary();
    }
  }

  /**
   * Requests that the lock for the given URI be released, interrupting the active stream and
   * creating a stop file to signal other replicas.
   */
  @Override
  public void requestLockRelease(String requestUri) {
    UploadId id = idFactory.readUploadId(requestUri);
    if (id == null) {
      return;
    }
    String idStr = id.toString();

    // 1. Release JVM-local lock if active
    WeakReference<InterruptibleInputStream> streamRef = activeLocks.get(idStr);
    if (streamRef != null) {
      InterruptibleInputStream stream = streamRef.get();
      if (stream != null) {
        stream.interrupt();
      }
      activeLocks.remove(idStr);
    }

    // 2. Create the stop file to signal other replicas
    Path stopFilePath = getStopPath(id);
    try {
      Path parentDir = stopFilePath.getParent();
      if (parentDir != null && !Files.exists(parentDir)) {
        Files.createDirectories(parentDir);
      }
      Files.write(stopFilePath, new byte[0]);
    } catch (IOException e) {
      log.warn("Unable to create stop file " + stopFilePath, e);
    }
  }

  /** Spawns a new background watchdog thread if none is currently active. */
  private void startWatchdogIfNecessary() {
    synchronized (watchdogLock) {
      if (watchdogThread == null || !watchdogThread.isAlive()) {
        watchdogThread = new Thread(new LockWatchdogRunnable(), "tus-lock-watchdog");
        watchdogThread.setDaemon(true);
        // Set lowest priority to ensure request threads are prioritized by the OS scheduler
        watchdogThread.setPriority(Thread.MIN_PRIORITY);
        watchdogThread.start();
      }
    }
  }

  private Path getLockPath(UploadId id) {
    return getPathInStorageDirectory(id);
  }

  /**
   * Resolves the stop file path based on the upload ID, ensuring files reside in the correct
   * directory.
   */
  private Path getStopPath(UploadId id) {
    Path lockPath = getPathInStorageDirectory(id);
    return lockPath.resolveSibling(id.toString() + ".stop");
  }

  /**
   * Runnable implementation for the background watchdog thread. This thread polls the storage
   * directory for ".stop" files created by other processes or replicas. If a stop file is found for
   * an active local upload, it interrupts the request stream to release the lock. The thread
   * self-terminates when there are no more active local locks to monitor.
   */
  private class LockWatchdogRunnable implements Runnable {
    @Override
    public void run() {
      try {
        while (true) {
          try {
            Thread.sleep(1000L);
          } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            break;
          }

          // Check stop files for each active lock
          for (Map.Entry<String, WeakReference<InterruptibleInputStream>> entry :
              activeLocks.entrySet()) {
            String idStr = entry.getKey();
            WeakReference<InterruptibleInputStream> ref = entry.getValue();
            InterruptibleInputStream stream = ref.get();

            if (stream == null) {
              activeLocks.remove(idStr);
              continue;
            }

            Path stopFilePath = getStopPath(new UploadId(idStr));
            if (stopFilePath != null && Files.exists(stopFilePath)) {
              try {
                log.info(
                    "Watchdog detected stop file for upload ID {}. Interrupting stream.", idStr);
                stream.interrupt();
              } catch (Throwable t) {
                log.warn("Error interrupting stream for ID " + idStr, t);
              }
              activeLocks.remove(idStr);
              try {
                Files.deleteIfExists(stopFilePath);
              } catch (IOException e) {
                // ignore
              }
            }
          }

          // Thread-safe check to decide whether to exit
          if (activeLocks.isEmpty()) {
            synchronized (watchdogLock) {
              if (activeLocks.isEmpty()) {
                watchdogThread = null;
                break;
              }
            }
          }
        }
      } catch (Throwable t) {
        log.error("Lock watchdog encountered an unexpected error", t);
        synchronized (watchdogLock) {
          watchdogThread = null;
        }
      }
    }
  }

  /**
   * Decorator around UploadLock to manage local map registry cleanup and stop file deletion when
   * the lock is released or closed.
   */
  private class RegisteredLock implements UploadLock {
    private final UploadLock delegate;
    private final String uploadIdStr;
    private final Path stopFilePath;

    public RegisteredLock(UploadLock delegate, String uploadIdStr, Path stopFilePath) {
      this.delegate = delegate;
      this.uploadIdStr = uploadIdStr;
      this.stopFilePath = stopFilePath;
    }

    @Override
    public String getUploadUri() {
      return delegate.getUploadUri();
    }

    @Override
    public void release() {
      try {
        delegate.release();
      } finally {
        activeLocks.remove(uploadIdStr);
        if (stopFilePath != null) {
          try {
            Files.deleteIfExists(stopFilePath);
          } catch (IOException e) {
            log.warn("Unable to delete stop file " + stopFilePath, e);
          }
        }
      }
    }

    @Override
    public void close() throws IOException {
      try {
        delegate.close();
      } finally {
        activeLocks.remove(uploadIdStr);
        if (stopFilePath != null) {
          try {
            Files.deleteIfExists(stopFilePath);
          } catch (IOException e) {
            log.warn("Unable to delete stop file " + stopFilePath, e);
          }
        }
      }
    }
  }
}

1	package me.desair.tus.server.upload.disk;
2
3	import java.io.File;
4	import java.io.IOException;
5	import java.io.InputStream;
6	import java.lang.ref.WeakReference;
7	import java.nio.file.DirectoryStream;
8	import java.nio.file.Files;
9	import java.nio.file.Path;
10	import java.nio.file.attribute.FileTime;
11	import java.util.Map;
12	import java.util.concurrent.ConcurrentHashMap;
13	import me.desair.tus.server.exception.TusException;
14	import me.desair.tus.server.exception.UploadAlreadyLockedException;
15	import me.desair.tus.server.upload.UploadId;
16	import me.desair.tus.server.upload.UploadIdFactory;
17	import me.desair.tus.server.upload.UploadLock;
18	import me.desair.tus.server.upload.UploadLockingService;
19	import me.desair.tus.server.util.InterruptibleInputStream;
20	import org.apache.commons.lang3.Validate;
21	import org.slf4j.Logger;
22	import org.slf4j.LoggerFactory;
23
24	/**
25	* {@link UploadLockingService} implementation that uses the file system for implementing locking
26	* <br>
27	* File locking can also apply to shared network drives. This way the framework supports clustering
28	* as long as the upload storage directory is mounted as a shared (network) drive. <br>
29	* File locks are also automatically released on application (JVM) shutdown. This means the file
30	* locking is not persistent and prevents cleanup and stale lock issues.
31	*/
32	public class DiskLockingService extends AbstractDiskBasedService implements UploadLockingService {
33
34	private static final Logger log = LoggerFactory.getLogger(DiskLockingService.class);	6✔
35	private static final String LOCK_SUB_DIRECTORY = "locks";
36
37	/** Registry tracking active request input streams in the current JVM. */
38	private static final ConcurrentHashMap<String, WeakReference<InterruptibleInputStream>>
39	activeLocks = new ConcurrentHashMap<>();	8✔
40
41	private static Thread watchdogThread = null;	4✔
42	private static final Object watchdogLock = new Object();	10✔
43
44	private UploadIdFactory idFactory;
45
46	public DiskLockingService(String storagePath) {
47	super(storagePath + File.separator + LOCK_SUB_DIRECTORY);	10✔
48	}	2✔
49
50	/** Constructor to use custom UploadIdFactory. */
51	public DiskLockingService(UploadIdFactory idFactory, String storagePath) {
52	this(storagePath);	6✔
53	Validate.notNull(idFactory, "The IdFactory cannot be null");	12✔
54	this.idFactory = idFactory;	6✔
55	}	2✔
56
57	/**
58	* Attempts to lock the upload resource. Wraps the lock in a RegisteredLock decorator to manage
59	* cleanup of stop files and the active lock registry.
60	*/
61	@Override
62	public UploadLock lockUploadByUri(String requestUri) throws TusException, IOException {
63
64	UploadId id = idFactory.readUploadId(requestUri);	10✔
65
66	UploadLock lock = null;	4✔
67
68	Path lockPath = getLockPath(id);	8✔
69	// If lockPath is not null, we know this is a valid Upload URI
70	if (lockPath != null) {	4✔
71	FileBasedLock baseLock = new FileBasedLock(requestUri, lockPath);	12✔
72	Path stopFilePath = baseLock.getLockPath().resolveSibling(id.toString() + ".stop");	14✔
73	lock = new RegisteredLock(baseLock, id.toString(), stopFilePath);	18✔
74	}
75	return lock;	4✔
76	}
77
78	/** Cleans up stale locks and stop files in the storage directory. */
79	@Override
80	public void cleanupStaleLocks() throws IOException {
81	try (DirectoryStream<Path> locksStream = Files.newDirectoryStream(getStoragePath())) {	8✔
82	for (Path path : locksStream) {	16✔
83	if (Files.exists(path)) {	5!
84	FileTime lastModifiedTime = Files.getLastModifiedTime(path);	5✔
85	if (lastModifiedTime.toMillis() < System.currentTimeMillis() - 10000L) {	7✔
86	String fileName = path.getFileName().toString();	4✔
87	if (fileName.endsWith(".stop")) {	4✔
88	Files.deleteIfExists(path);	4✔
89	} else {
90	UploadId id = new UploadId(fileName);	5✔
91	if (!isLocked(id)) {	4✔
92	Files.deleteIfExists(path);	3✔
93	Files.deleteIfExists(path.resolveSibling(fileName + ".stop"));	6✔
94	}
95	}
96	}
97	}
98	}	1✔
99	}
100	}	2✔
101
102	/** Checks whether the upload is locked by attempting to obtain a short-lived file lock. */
103	@Override
104	public boolean isLocked(UploadId id) {
105	boolean locked = false;	4✔
106	Path lockPath = getLockPath(id);	8✔
107
108	if (lockPath != null) {	4!
109	// Try to obtain a lock to see if the upload is currently locked
110	try (UploadLock lock = new FileBasedLock(id.toString(), lockPath)) {	14✔
111
112	// We got the lock, so it means no one else is locking it.
113	locked = false;	4✔
114
115	} catch (UploadAlreadyLockedException \| IOException e) {	1✔
116	// There was already a lock
117	locked = true;	2✔
118	}	2✔
119	}
120
121	return locked;	4✔
122	}
123
124	@Override
125	public void setIdFactory(UploadIdFactory idFactory) {
126	Validate.notNull(idFactory, "The IdFactory cannot be null");	12✔
127	this.idFactory = idFactory;	6✔
128	}	2✔
129
130	/**
131	* Registers an active request input stream so that it can be interrupted if lock contention
132	* occurs.
133	*/
134	@Override
135	public void registerInputStream(String requestUri, InputStream inputStream) {
136	if (inputStream == null) {	4✔
137	return;	1✔
138	}
139	UploadId id = idFactory.readUploadId(requestUri);	10✔
140	if (id == null) {	4✔
141	return;	1✔
142	}
143	if (inputStream instanceof InterruptibleInputStream) {	6!
144	activeLocks.put(id.toString(), new WeakReference<>((InterruptibleInputStream) inputStream));	20✔
145	startWatchdogIfNecessary();	4✔
146	}
147	}	2✔
148
149	/**
150	* Requests that the lock for the given URI be released, interrupting the active stream and
151	* creating a stop file to signal other replicas.
152	*/
153	@Override
154	public void requestLockRelease(String requestUri) {
155	UploadId id = idFactory.readUploadId(requestUri);	10✔
156	if (id == null) {	4✔
157	return;	1✔
158	}
159	String idStr = id.toString();	6✔
160
161	// 1. Release JVM-local lock if active
162	WeakReference<InterruptibleInputStream> streamRef = activeLocks.get(idStr);	10✔
163	if (streamRef != null) {	4✔
164	InterruptibleInputStream stream = streamRef.get();	8✔
165	if (stream != null) {	4✔
166	stream.interrupt();	4✔
167	}
168	activeLocks.remove(idStr);	8✔
169	}
170
171	// 2. Create the stop file to signal other replicas
172	Path stopFilePath = getStopPath(id);	8✔
173	try {
174	Path parentDir = stopFilePath.getParent();	6✔
175	if (parentDir != null && !Files.exists(parentDir)) {	14!
176	Files.createDirectories(parentDir);	5✔
177	}
178	Files.write(stopFilePath, new byte[0]);	14✔
179	} catch (IOException e) {	1✔
180	log.warn("Unable to create stop file " + stopFilePath, e);	6✔
181	}	2✔
182	}	2✔
183
184	/** Spawns a new background watchdog thread if none is currently active. */
185	private void startWatchdogIfNecessary() {
186	synchronized (watchdogLock) {	8✔
187	if (watchdogThread == null \|\| !watchdogThread.isAlive()) {	10✔
188	watchdogThread = new Thread(new LockWatchdogRunnable(), "tus-lock-watchdog");	18✔
189	watchdogThread.setDaemon(true);	6✔
190	// Set lowest priority to ensure request threads are prioritized by the OS scheduler
191	watchdogThread.setPriority(Thread.MIN_PRIORITY);	6✔
192	watchdogThread.start();	4✔
193	}
194	}	6✔
195	}	2✔
196
197	private Path getLockPath(UploadId id) {
198	return getPathInStorageDirectory(id);	8✔
199	}
200
201	/**
202	* Resolves the stop file path based on the upload ID, ensuring files reside in the correct
203	* directory.
204	*/
205	private Path getStopPath(UploadId id) {
206	Path lockPath = getPathInStorageDirectory(id);	8✔
207	return lockPath.resolveSibling(id.toString() + ".stop");	12✔
208	}
209
210	/**
211	* Runnable implementation for the background watchdog thread. This thread polls the storage
212	* directory for ".stop" files created by other processes or replicas. If a stop file is found for
213	* an active local upload, it interrupts the request stream to release the lock. The thread
214	* self-terminates when there are no more active local locks to monitor.
215	*/
216	private class LockWatchdogRunnable implements Runnable {	12✔
217	@Override
218	public void run() {
219	try {
220	while (true) {
221	try {
222	Thread.sleep(1000L);	4✔
223	} catch (InterruptedException e) {	1✔
224	Thread.currentThread().interrupt();	2✔
225	break;	1✔
226	}	2✔
227
228	// Check stop files for each active lock
229	for (Map.Entry<String, WeakReference<InterruptibleInputStream>> entry :
230	activeLocks.entrySet()) {	18✔
231	String idStr = entry.getKey();	4✔
232	WeakReference<InterruptibleInputStream> ref = entry.getValue();	4✔
233	InterruptibleInputStream stream = ref.get();	4✔
234
235	if (stream == null) {	2✔
236	activeLocks.remove(idStr);	4✔
237	continue;	1✔
238	}
239
240	Path stopFilePath = getStopPath(new UploadId(idStr));	8✔
241	if (stopFilePath != null && Files.exists(stopFilePath)) {	7!
242	try {
243	log.info(	4✔
244	"Watchdog detected stop file for upload ID {}. Interrupting stream.", idStr);
245	stream.interrupt();	2✔
246	} catch (Throwable t) {	1✔
247	log.warn("Error interrupting stream for ID " + idStr, t);	5✔
248	}	1✔
249	activeLocks.remove(idStr);	4✔
250	try {
251	Files.deleteIfExists(stopFilePath);	3✔
NEW 252	} catch (IOException e) {	×
253	// ignore
254	}	1✔
255	}
256	}	1✔
257
258	// Thread-safe check to decide whether to exit
259	if (activeLocks.isEmpty()) {	6!
260	synchronized (watchdogLock) {	8✔
261	if (activeLocks.isEmpty()) {	6!
262	watchdogThread = null;	4✔
263	break;	6✔
264	}
NEW 265	}	×
266	}
267	}
268	} catch (Throwable t) {	1✔
269	log.error("Lock watchdog encountered an unexpected error", t);	4✔
270	synchronized (watchdogLock) {	4✔
271	watchdogThread = null;	2✔
272	}	3✔
273	}	2✔
274	}	2✔
275	}
276
277	/**
278	* Decorator around UploadLock to manage local map registry cleanup and stop file deletion when
279	* the lock is released or closed.
280	*/
281	private class RegisteredLock implements UploadLock {
282	private final UploadLock delegate;
283	private final String uploadIdStr;
284	private final Path stopFilePath;
285
286	public RegisteredLock(UploadLock delegate, String uploadIdStr, Path stopFilePath) {	10✔
287	this.delegate = delegate;	6✔
288	this.uploadIdStr = uploadIdStr;	6✔
289	this.stopFilePath = stopFilePath;	6✔
290	}	2✔
291
292	@Override
293	public String getUploadUri() {
294	return delegate.getUploadUri();	4✔
295	}
296
297	@Override
298	public void release() {
299	try {
300	delegate.release();	3✔
301	} finally {
302	activeLocks.remove(uploadIdStr);	5✔
303	if (stopFilePath != null) {	3!
304	try {
305	Files.deleteIfExists(stopFilePath);	4✔
306	} catch (IOException e) {	1✔
307	log.warn("Unable to delete stop file " + stopFilePath, e);	7✔
308	}	1✔
309	}
310	}
311	}	1✔
312
313	@Override
314	public void close() throws IOException {
315	try {
316	delegate.close();	6✔
317	} finally {
318	activeLocks.remove(uploadIdStr);	10✔
319	if (stopFilePath != null) {	6!
320	try {
321	Files.deleteIfExists(stopFilePath);	8✔
322	} catch (IOException e) {	1✔
323	log.warn("Unable to delete stop file " + stopFilePath, e);	7✔
324	}	2✔
325	}
326	}
327	}	2✔
328	}
329	}

tomdesair / tus-java-server / 27470493850

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous