28397983898

Committed 29 Jun 2026 07:41PM UTC coverage: 95.068% (-0.06%) from 95.129%

Build # 28397983898

Build Type

push

github

Committed by

web-flow

Commit Message

Refactor `LockWatchdogRunnable.run()` into smaller helper methods (#101)

Extracted lock cleanup logic within `DiskLockingService` into discrete helper methods (`checkActiveLocks`, `checkStopFileAndInterrupt`, `shouldExitWatchdog`) to improve code readability and maintainability. Functionality remains strictly identical.

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
Co-authored-by: tomdesair <14034630+tomdesair@users.noreply.github.com>

Coverage Stats

623 of 702 branches covered (88.75%)

Branch coverage included in aggregate %.

29 of 32 new or added lines in 1 file covered. (90.63%)

1825 of 1873 relevant lines covered (97.44%)

6.58 hits per line

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

93.56

/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.Objects;
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.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);
    Objects.requireNonNull(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) {
    Objects.requireNonNull(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;
          }

          checkActiveLocks();

          if (shouldExitWatchdog()) {
            break;
          }
        }
      } catch (Throwable t) {
        log.error("Lock watchdog encountered an unexpected error", t);
        synchronized (watchdogLock) {
          watchdogThread = null;
        }
      }
    }

    private void checkActiveLocks() {
      // 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;
        }

        checkStopFileAndInterrupt(idStr, stream);
      }
    }

    private void checkStopFileAndInterrupt(String idStr, InterruptibleInputStream stream) {
      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
        }
      }
    }

    private boolean shouldExitWatchdog() {
      // Thread-safe check to decide whether to exit
      if (activeLocks.isEmpty()) {
        synchronized (watchdogLock) {
          if (activeLocks.isEmpty()) {
            watchdogThread = null;
            return true;
          }
        }
      }
      return false;
    }
  }

  /**
   * 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.Objects;
13	import java.util.concurrent.ConcurrentHashMap;
14	import me.desair.tus.server.exception.TusException;
15	import me.desair.tus.server.exception.UploadAlreadyLockedException;
16	import me.desair.tus.server.upload.UploadId;
17	import me.desair.tus.server.upload.UploadIdFactory;
18	import me.desair.tus.server.upload.UploadLock;
19	import me.desair.tus.server.upload.UploadLockingService;
20	import me.desair.tus.server.util.InterruptibleInputStream;
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	Objects.requireNonNull(idFactory, "The IdFactory cannot be null");	8✔
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	Objects.requireNonNull(idFactory, "The IdFactory cannot be null");	8✔
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	checkActiveLocks();	4✔
229
230	if (shouldExitWatchdog()) {	6!
231	break;	2✔
232	}
233	}
234	} catch (Throwable t) {	1✔
235	log.error("Lock watchdog encountered an unexpected error", t);	4✔
236	synchronized (watchdogLock) {	4✔
237	watchdogThread = null;	2✔
238	}	3✔
239	}	2✔
240	}	2✔
241
242	private void checkActiveLocks() {
243	// Check stop files for each active lock
244	for (Map.Entry<String, WeakReference<InterruptibleInputStream>> entry :
245	activeLocks.entrySet()) {	18✔
246	String idStr = entry.getKey();	4✔
247	WeakReference<InterruptibleInputStream> ref = entry.getValue();	4✔
248	InterruptibleInputStream stream = ref.get();	4✔
249
250	if (stream == null) {	2✔
251	activeLocks.remove(idStr);	4✔
252	continue;	1✔
253	}
254
255	checkStopFileAndInterrupt(idStr, stream);	4✔
256	}	1✔
257	}	2✔
258
259	private void checkStopFileAndInterrupt(String idStr, InterruptibleInputStream stream) {
260	Path stopFilePath = getStopPath(new UploadId(idStr));	8✔
261	if (stopFilePath != null && Files.exists(stopFilePath)) {	7!
262	try {
263	log.info("Watchdog detected stop file for upload ID {}. Interrupting stream.", idStr);	4✔
264	stream.interrupt();	2✔
265	} catch (Throwable t) {	1✔
266	log.warn("Error interrupting stream for ID " + idStr, t);	5✔
267	}	1✔
268	activeLocks.remove(idStr);	4✔
269	try {
270	Files.deleteIfExists(stopFilePath);	3✔
NEW 271	} catch (IOException e) {	×
272	// ignore
273	}	1✔
274	}
275	}	1✔
276
277	private boolean shouldExitWatchdog() {
278	// Thread-safe check to decide whether to exit
279	if (activeLocks.isEmpty()) {	6!
280	synchronized (watchdogLock) {	8✔
281	if (activeLocks.isEmpty()) {	6!
282	watchdogThread = null;	4✔
283	return true;	8✔
284	}
NEW 285	}	×
286	}
NEW 287	return false;	×
288	}
289	}
290
291	/**
292	* Decorator around UploadLock to manage local map registry cleanup and stop file deletion when
293	* the lock is released or closed.
294	*/
295	private class RegisteredLock implements UploadLock {
296	private final UploadLock delegate;
297	private final String uploadIdStr;
298	private final Path stopFilePath;
299
300	public RegisteredLock(UploadLock delegate, String uploadIdStr, Path stopFilePath) {	10✔
301	this.delegate = delegate;	6✔
302	this.uploadIdStr = uploadIdStr;	6✔
303	this.stopFilePath = stopFilePath;	6✔
304	}	2✔
305
306	@Override
307	public String getUploadUri() {
308	return delegate.getUploadUri();	4✔
309	}
310
311	@Override
312	public void release() {
313	try {
314	delegate.release();	3✔
315	} finally {
316	activeLocks.remove(uploadIdStr);	5✔
317	if (stopFilePath != null) {	3!
318	try {
319	Files.deleteIfExists(stopFilePath);	4✔
320	} catch (IOException e) {	1✔
321	log.warn("Unable to delete stop file " + stopFilePath, e);	7✔
322	}	1✔
323	}
324	}
325	}	1✔
326
327	@Override
328	public void close() throws IOException {
329	try {
330	delegate.close();	6✔
331	} finally {
332	activeLocks.remove(uploadIdStr);	10✔
333	if (stopFilePath != null) {	6!
334	try {
335	Files.deleteIfExists(stopFilePath);	8✔
336	} catch (IOException e) {	1✔
337	log.warn("Unable to delete stop file " + stopFilePath, e);	7✔
338	}	2✔
339	}
340	}
341	}	2✔
342	}
343	}

tomdesair / tus-java-server / 28397983898

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