fix: hold upload lock during stream interruption until data is persisted

.gitignore

@@ -179,3 +179,4 @@ AZURE_BLOB_STORAGE_IMPROVEMENTS.md
 SFTP_STORAGE_ANALYSIS.md
 NFS_LOCKING.md
 RELEASE-REVIEW.md
+ASYNC_REVIEW.md

src/main/java/me/desair/tus/server/TusFileUploadService.java

@@ -533,7 +533,6 @@ public UploadInfo process(
     boolean wasInProgress = true;
 
     try (UploadLock lock = acquireUploadLock(method, request.getRequestURI())) {
-      request.setUploadLock(lock);
       // Evaluate initial progress state safely under the acquired upload lock
       // to avoid uncoordinated disk I/O or race conditions with concurrent requests.
       wasInProgress = checkWasInProgress(request, ownerKey);

src/main/java/me/desair/tus/server/core/CorePatchRequestHandler.java

@@ -72,8 +72,7 @@ public HttpProblemDetails process(
         // If a locking service is provided, wrap the input stream in an InterruptibleInputStream
         // and register it with the locking service to allow for interruption of the upload.
         if (lockingService != null) {
-          InterruptibleInputStream interruptibleStream =
-              new InterruptibleInputStream(stream, servletRequest.getUploadLock());
+          InterruptibleInputStream interruptibleStream = new InterruptibleInputStream(stream);
           lockingService.registerInputStream(servletRequest.getRequestURI(), interruptibleStream);
           stream = interruptibleStream;
         }

src/main/java/me/desair/tus/server/creationwithupload/CreationWithUploadPostRequestHandler.java

@@ -47,8 +47,7 @@ public HttpProblemDetails process(
         if (uploadInfo != null && uploadInfo.isUploadInProgress()) {
           InputStream stream = servletRequest.getContentInputStream();
           if (uploadLockingService != null) {
-            InterruptibleInputStream interruptibleStream =
-                new InterruptibleInputStream(stream, servletRequest.getUploadLock());
+            InterruptibleInputStream interruptibleStream = new InterruptibleInputStream(stream);
             uploadLockingService.registerInputStream(location, interruptibleStream);
             stream = interruptibleStream;
           }

src/main/java/me/desair/tus/server/rufh/handler/RufhAppendPatchRequestHandler.java

@@ -58,8 +58,7 @@ public HttpProblemDetails process(
     InputStream is = servletRequest.getContentInputStream();
     if (is != null) {
       if (uploadLockingService != null) {
-        InterruptibleInputStream interruptibleStream =
-            new InterruptibleInputStream(is, servletRequest.getUploadLock());
+        InterruptibleInputStream interruptibleStream = new InterruptibleInputStream(is);
         uploadLockingService.registerInputStream(requestUri, interruptibleStream);
         is = interruptibleStream;
       }

src/main/java/me/desair/tus/server/rufh/handler/RufhCreationPostRequestHandler.java

@@ -87,8 +87,7 @@ public HttpProblemDetails process(
     InputStream is = servletRequest.getContentInputStream();
     if (is != null && servletRequest.getContentLengthLong() != 0) {
       if (uploadLockingService != null) {
-        InterruptibleInputStream interruptibleStream =
-            new InterruptibleInputStream(is, servletRequest.getUploadLock());
+        InterruptibleInputStream interruptibleStream = new InterruptibleInputStream(is);
         uploadLockingService.registerInputStream(uploadUri, interruptibleStream);
         is = interruptibleStream;
       }

src/main/java/me/desair/tus/server/util/InterruptibleInputStream.java

@@ -2,7 +2,6 @@
 
 import java.io.IOException;
 import java.io.InputStream;
-import me.desair.tus.server.upload.UploadLock;
 
 /**
  * An InputStream wrapper that can be interrupted by another thread. When interrupted, it throws an
@@ -11,51 +10,19 @@
 public class InterruptibleInputStream extends InputStream {
 
   private final InputStream delegate;
-  private UploadLock uploadLock;
   private volatile boolean interrupted = false;
 
   /**
-   * Constructs an interruptible input stream wrapping the given delegate stream and associating it
-   * with the provided {@link UploadLock}.
+   * Constructs an interruptible input stream wrapping the given delegate stream.
    *
    * @param delegate The delegate input stream to wrap
-   * @param uploadLock The upload lock associated with this stream, or null
    * @throws IllegalArgumentException if the delegate is null
    */
-  public InterruptibleInputStream(InputStream delegate, UploadLock uploadLock) {
+  public InterruptibleInputStream(InputStream delegate) {
     if (delegate == null) {
       throw new IllegalArgumentException("Delegate InputStream cannot be null");
     }
     this.delegate = delegate;
-    this.uploadLock = uploadLock;
-  }
-
-  /**
-   * Constructs an interruptible input stream wrapping the given delegate stream.
-   *
-   * @param delegate The delegate input stream to wrap
-   * @throws IllegalArgumentException if the delegate is null
-   */
-  public InterruptibleInputStream(InputStream delegate) {
-    this(delegate, null);
-  }
-
-  /**
-   * Sets the corresponding {@link UploadLock} associated with this stream.
-   *
-   * @param uploadLock The upload lock
-   */
-  public void setCorrespondingUploadLock(UploadLock uploadLock) {
-    this.uploadLock = uploadLock;
-  }
-
-  /**
-   * Retrieves the corresponding {@link UploadLock} associated with this stream.
-   *
-   * @return The upload lock, or null if none is set
-   */
-  public UploadLock getCorrespondingUploadLock() {
-    return uploadLock;
   }
 
   private void checkInterrupted() throws IOException {
@@ -64,16 +31,24 @@ private void checkInterrupted() throws IOException {
     }
   }
 
+  /**
+   * Interrupts the active stream by marking it interrupted and closing the underlying delegate
+   * stream to abort any blocking socket read operations.
+   *
+   * <p>Note: Stream interruption intentionally does NOT release the upload lock. The thread
+   * processing the upload must continue holding the lock while it persists any buffered or received
+   * payload bytes to storage (e.g. S3, Azure Blob, or Disk) and updates the metadata offset. The
+   * upload lock is released only when the upload handler finishes processing and exits its
+   * try-with-resources lock block. This guarantees that concurrent or resuming requests (such as
+   * HEAD requests) wait for all in-flight bytes to be committed before reading the upload offset.
+   */
   public void interrupt() {
     interrupted = true;
     try {
       delegate.close();
     } catch (IOException e) {
       // Ignore close exception during interrupt
     }
-    if (uploadLock != null) {
-      uploadLock.release();
-    }
   }
 
   public boolean isInterrupted() {

src/main/java/me/desair/tus/server/util/TusServletRequest.java

@@ -16,15 +16,13 @@
 import me.desair.tus.server.HttpHeader;
 import me.desair.tus.server.TusExtension;
 import me.desair.tus.server.checksum.ChecksumAlgorithm;
-import me.desair.tus.server.upload.UploadLock;
 import org.apache.commons.codec.binary.Base64;
 import org.apache.commons.io.input.BoundedInputStream;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.Strings;
 
 public class TusServletRequest extends HttpServletRequestWrapper {
 
-  private UploadLock uploadLock;
   private BoundedInputStream countingInputStream;
   private Map<ChecksumAlgorithm, DigestInputStream> digestInputStreamMap =
       new EnumMap<>(ChecksumAlgorithm.class);
@@ -152,24 +150,6 @@ private boolean hasChunkedTransferEncoding() {
     return Strings.CI.equals("chunked", getHeader(HttpHeader.TRANSFER_ENCODING));
   }
 
-  /**
-   * Retrieves the active {@link UploadLock} associated with this request.
-   *
-   * @return The active upload lock, or null if no lock is held
-   */
-  public UploadLock getUploadLock() {
-    return uploadLock;
-  }
-
-  /**
-   * Sets the active {@link UploadLock} associated with this request.
-   *
-   * @param uploadLock The active upload lock
-   */
-  public void setUploadLock(UploadLock uploadLock) {
-    this.uploadLock = uploadLock;
-  }
-
   private MessageDigest getMessageDigest(ChecksumAlgorithm algorithm) {
     if (digestInputStreamMap.containsKey(algorithm)) {
       return digestInputStreamMap.get(algorithm).getMessageDigest();

src/test/java/me/desair/tus/server/AbstractITTusFileUploadService.java

@@ -16,6 +16,7 @@
 import static org.junit.Assert.fail;
 
 import jakarta.servlet.http.HttpServletResponse;
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
@@ -1694,6 +1695,180 @@ public void run() {
     bgThread.join(2000);
   }
 
+  /**
+   * Tests that when an upload stream is interrupted/paused midway after receiving partial bytes,
+   * the thread performing the upload continues to hold the lock while persisting all received bytes
+   * to storage (S3, Azure Blob, Disk) and updating the metadata offset. A concurrent HEAD request
+   * (resuming the upload) waits for the upload thread to complete persistence and release the lock,
+   * receiving the up-to-date offset without rollback or data loss.
+   */
+  @Test
+  public void testInterruptedUploadPersistsBytesBeforeHeadResponds() throws Exception {
+    // 1. Create upload resource for 1000 bytes
+    servletRequest.setMethod("POST");
+    servletRequest.setRequestURI(UPLOAD_URI);
+    servletRequest.addHeader(HttpHeader.CONTENT_LENGTH, 0);
+    servletRequest.addHeader(HttpHeader.UPLOAD_LENGTH, 1000L);
+    servletRequest.addHeader(HttpHeader.TUS_RESUMABLE, "1.0.0");
+    tusFileUploadService.process(servletRequest, servletResponse, OWNER_KEY);
+    String location =
+        UPLOAD_URI
+            + StringUtils.substringAfter(
+                servletResponse.getHeader(HttpHeader.LOCATION), UPLOAD_URI);
+
+    // 2. Start a PATCH in a background thread that streams 100 bytes and then blocks
+    final byte[] payload = new byte[100];
+    java.util.Arrays.fill(payload, (byte) 'A');
+    final java.util.concurrent.CountDownLatch bytesDelivered =
+        new java.util.concurrent.CountDownLatch(1);
+    final java.util.concurrent.atomic.AtomicReference<Exception> bgException =
+        new java.util.concurrent.atomic.AtomicReference<>();
+
+    InputStream partialBlockingStream =
+        new InputStream() {
+          private final InputStream byteStream = new ByteArrayInputStream(payload);
+          private volatile boolean closed = false;
+
+          @Override
+          public int read() throws IOException {
+            int b = byteStream.read();
+            if (b != -1) {
+              return b;
+            }
+            // All 100 payload bytes delivered; notify that upload thread has buffered them
+            bytesDelivered.countDown();
+            synchronized (this) {
+              while (!closed) {
+                try {
+                  this.wait(100);
+                } catch (InterruptedException e) {
+                  Thread.currentThread().interrupt();
+                  throw new IOException("Stream read interrupted", e);
+                }
+              }
+            }
+            throw new IOException("Stream closed");
+          }
+
+          @Override
+          public int read(byte[] b, int off, int len) throws IOException {
+            int read = byteStream.read(b, off, len);
+            if (read != -1) {
+              return read;
+            }
+            bytesDelivered.countDown();
+            synchronized (this) {
+              while (!closed) {
+                try {
+                  this.wait(100);
+                } catch (InterruptedException e) {
+                  Thread.currentThread().interrupt();
+                  throw new IOException("Stream read interrupted", e);
+                }
+              }
+            }
+            throw new IOException("Stream closed");
+          }
+
+          @Override
+          public void close() throws IOException {
+            synchronized (this) {
+              closed = true;
+              this.notifyAll();
+            }
+          }
+        };
+
+    final MockHttpServletRequest bgRequest =
+        new MockHttpServletRequest() {
+          @Override
+          public jakarta.servlet.ServletInputStream getInputStream() {
+            return new jakarta.servlet.ServletInputStream() {
+              @Override
+              public int read() throws IOException {
+                return partialBlockingStream.read();
+              }
+
+              @Override
+              public int read(byte[] b, int off, int len) throws IOException {
+                return partialBlockingStream.read(b, off, len);
+              }
+
+              @Override
+              public void close() throws IOException {
+                partialBlockingStream.close();
+              }
+
+              @Override
+              public boolean isFinished() {
+                return false;
+              }
+
+              @Override
+              public boolean isReady() {
+                return true;
+              }
+
+              @Override
+              public void setReadListener(jakarta.servlet.ReadListener readListener) {}
+            };
+          }
+        };
+    bgRequest.setMethod("PATCH");
+    bgRequest.setRequestURI(location);
+    bgRequest.addHeader(HttpHeader.CONTENT_TYPE, "application/offset+octet-stream");
+    bgRequest.addHeader(HttpHeader.CONTENT_LENGTH, 100);
+    bgRequest.addHeader(HttpHeader.UPLOAD_OFFSET, 0);
+    bgRequest.addHeader(HttpHeader.TUS_RESUMABLE, "1.0.0");
+
+    Thread bgThread =
+        new Thread(
+            () -> {
+              try {
+                MockHttpServletResponse bgResponse = new MockHttpServletResponse();
+                tusFileUploadService.process(bgRequest, bgResponse, OWNER_KEY);
+              } catch (Exception e) {
+                bgException.set(e);
+              }
+            });
+    bgThread.start();
+
+    // Wait for the background thread to read all 100 bytes and enter blocking wait
+    assertTrue(bytesDelivered.await(5, java.util.concurrent.TimeUnit.SECONDS));
+
+    // 3. Concurrent HEAD request (resume request) arrives within the same second
+    MockHttpServletRequest headRequest = new MockHttpServletRequest();
+    MockHttpServletResponse headResponse = new MockHttpServletResponse();
+    headRequest.setMethod("HEAD");
+    headRequest.setRequestURI(location);
+    headRequest.addHeader(HttpHeader.TUS_RESUMABLE, "1.0.0");
+
+    tusFileUploadService.process(headRequest, headResponse, OWNER_KEY);
+
+    // The HEAD request must succeed (204) and see the newly persisted offset of 100 bytes!
+    assertThat(headResponse.getStatus(), is(204));
+    assertThat(headResponse.getHeader(HttpHeader.UPLOAD_OFFSET), is("100"));
+
+    bgThread.join(3000);
+
+    // 4. Resume upload: send remaining 900 bytes starting at offset 100
+    MockHttpServletRequest resumeRequest = new MockHttpServletRequest();
+    MockHttpServletResponse resumeResponse = new MockHttpServletResponse();
+    resumeRequest.setMethod("PATCH");
+    resumeRequest.setRequestURI(location);
+    resumeRequest.addHeader(HttpHeader.CONTENT_TYPE, "application/offset+octet-stream");
+    resumeRequest.addHeader(HttpHeader.CONTENT_LENGTH, 900);
+    resumeRequest.addHeader(HttpHeader.UPLOAD_OFFSET, 100);
+    resumeRequest.addHeader(HttpHeader.TUS_RESUMABLE, "1.0.0");
+    byte[] remainingPayload = new byte[900];
+    java.util.Arrays.fill(remainingPayload, (byte) 'B');
+    resumeRequest.setContent(remainingPayload);
+
+    tusFileUploadService.process(resumeRequest, resumeResponse, OWNER_KEY);
+    assertThat(resumeResponse.getStatus(), is(204));
+    assertThat(resumeResponse.getHeader(HttpHeader.UPLOAD_OFFSET), is("1000"));
+  }
+
   @Test
   public void testCreationWithUploadOptions() throws Exception {
     reset();