blockIds,
+ long currentTotalAppended)
+ throws IOException, MaxAppendSizeExceededException {
+ long additionalAppended = 0L;
+ boolean streamFinished = false;
+
+ while (!streamFinished) {
+ File chunkFile = File.createTempFile("tus-azure-chunk-", ".tmp", tempBufferDir.toFile());
+ try {
+ long chunkSize = readChunk(inputStream, chunkFile, optimalBlockSize);
+ if (chunkSize <= 0) {
+ break;
+ }
+
+ additionalAppended += chunkSize;
+ long totalAppendedSoFar = currentTotalAppended + additionalAppended;
+ validateMaxAppendSize(totalAppendedSoFar, effectiveMaxAppendSize);
+
+ long currentOffset = upload.getOffset() + totalAppendedSoFar;
+ boolean complete = upload.getLength() != null && currentOffset == upload.getLength();
+
+ if (chunkSize < optimalBlockSize && !complete) {
+ bufferToPartBlob(partBlob, 0L, chunkFile, chunkSize);
+ } else {
+ stageChunkFile(chunkFile, chunkSize, blockBlobClient, blockIds);
+ }
+ } finally {
+ deleteFileQuietly(chunkFile);
+ }
+ }
+ return additionalAppended;
+ }
+
+ /** Buffers incoming data to temporary .part blob when under block threshold. */
+ private void bufferToPartBlob(
+ BlobClient partBlob, long existingPartSize, File tempFile, long appendSize)
+ throws IOException {
+ if (existingPartSize == 0) {
+ partBlob.upload(BinaryData.fromFile(tempFile.toPath()), true);
+ } else {
+ File combinedTemp = File.createTempFile("tus-azure-part-", ".tmp", tempBufferDir.toFile());
+ try {
+ try (InputStream partIs = partBlob.openInputStream();
+ InputStream tempIs = new FileInputStream(tempFile);
+ SequenceInputStream seqIs = new SequenceInputStream(partIs, tempIs);
+ FileOutputStream fos = new FileOutputStream(combinedTemp)) {
+ IOUtils.copyLarge(seqIs, fos);
+ }
+ partBlob.upload(BinaryData.fromFile(combinedTemp.toPath()), true);
+ } finally {
+ deleteFileQuietly(combinedTemp);
+ }
+ }
+ }
+
+ /** Commits empty block list for 0-byte upload completion. */
+ private void commitEmptyDataBlob(UploadId id) {
+ BlockBlobClient blockBlobClient =
+ containerClient.getBlobClient(uploadPrefix + id).getBlockBlobClient();
+ blockBlobClient.commitBlockList(new ArrayList<>(), true);
+ }
+
+ /** Generates Base64 encoded block ID matching Azure Block Blob standards. */
+ private String generateBlockId(int index) {
+ String idString = String.format("block-%06d", index);
+ return Base64.getEncoder().encodeToString(idString.getBytes(StandardCharsets.UTF_8));
+ }
+
+ /** Checks for deduplication match and links upload if duplicate. */
+ private void checkAndApplyDeduplication(UploadInfo uploadInfo) throws IOException {
+ if (!isUploadDeduplicationEnabled()
+ || uploadInfo == null
+ || uploadInfo.getChecksum() == null
+ || uploadInfo.getChecksumAlgorithm() == null) {
+ return;
+ }
+
+ String checksum = uploadInfo.getChecksum();
+ ChecksumAlgorithm algorithm = uploadInfo.getChecksumAlgorithm();
+
+ UploadInfo existingUpload = getUploadInfoByChecksum(checksum, algorithm);
+ if (existingUpload != null && !existingUpload.getId().equals(uploadInfo.getId())) {
+ log.info(
+ "Found duplicate upload with checksum {}/{} (parent ID {})",
+ algorithm,
+ checksum,
+ existingUpload.getId());
+ uploadInfo.setDuplicatesUploadId(existingUpload.getId());
+ containerClient.getBlobClient(uploadPrefix + uploadInfo.getId()).deleteIfExists();
+ } else {
+ String checksumKey = buildChecksumKey(checksum, algorithm);
+ BlobClient checksumBlob = containerClient.getBlobClient(checksumKey);
+ byte[] idBytes = uploadInfo.getId().toString().getBytes(StandardCharsets.UTF_8);
+ checksumBlob.upload(BinaryData.fromBytes(idBytes), true);
+ }
+ }
+
+ /** Builds checksum index key path. */
+ private String buildChecksumKey(String checksum, ChecksumAlgorithm algorithm) {
+ String algorithmName = algorithm != null ? algorithm.getTusName().toLowerCase() : "unknown";
+ return checksumsPrefix + algorithmName + "/" + checksum;
+ }
+
+ /** Sanitizes object key prefix format. */
+ private String sanitizePrefix(String prefix) {
+ if (prefix == null || prefix.isEmpty()) {
+ return "";
+ }
+ String result = prefix.startsWith("/") ? prefix.substring(1) : prefix;
+ return result.endsWith("/") ? result : result + "/";
+ }
+
+ /** Deletes local file suppressing exceptions. */
+ private void deleteFileQuietly(File file) {
+ if (file != null && file.exists()) {
+ try {
+ Files.delete(file.toPath());
+ } catch (Exception ignored) {
+ }
+ }
+ }
+
+ /** Ensures local directory exists. */
+ private void ensureDirectoryExists(Path dir) {
+ try {
+ if (!Files.exists(dir)) {
+ Files.createDirectories(dir);
+ }
+ } catch (IOException e) {
+ throw new RuntimeException("Could not create buffer directory " + dir, e);
+ }
+ }
+}
diff --git a/src/main/java/me/desair/tus/server/upload/azure/AzureBlobUploadLock.java b/src/main/java/me/desair/tus/server/upload/azure/AzureBlobUploadLock.java
new file mode 100644
index 0000000..e5ae9bb
--- /dev/null
+++ b/src/main/java/me/desair/tus/server/upload/azure/AzureBlobUploadLock.java
@@ -0,0 +1,103 @@
+package me.desair.tus.server.upload.azure;
+
+import com.azure.storage.blob.BlobClient;
+import com.azure.storage.blob.specialized.BlobLeaseClient;
+import java.io.IOException;
+import java.util.Objects;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import me.desair.tus.server.upload.UploadLock;
+import me.desair.tus.server.util.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Distributed upload lock implementation backed by Azure Blob Storage Leases.
+ *
+ * Azure Blob Leases provide native, atomic distributed locks. This class wraps an active lease
+ * and maintains a background daemon executor that periodically renews the lease (every 10 seconds
+ * for a standard 30-second lease) to prevent lock expiry during long-running streaming upload
+ * operations.
+ */
+public class AzureBlobUploadLock implements UploadLock {
+
+ private static final Logger log = LoggerFactory.getLogger(AzureBlobUploadLock.class);
+
+ private static final long RENEWAL_INTERVAL_SECONDS = 10L;
+
+ private final BlobLeaseClient leaseClient;
+ private final BlobClient lockBlob;
+ private final String uploadUri;
+ private final ScheduledExecutorService renewalExecutor;
+ private volatile boolean released = false;
+
+ /**
+ * Constructs an {@link AzureBlobUploadLock} wrapping an acquired Azure Blob lease.
+ *
+ * @param leaseClient The pre-acquired {@link BlobLeaseClient} holding the lease
+ * @param lockBlob The target lock {@link BlobClient}
+ * @param uploadUri The upload URI associated with this lock
+ */
+ public AzureBlobUploadLock(BlobLeaseClient leaseClient, BlobClient lockBlob, String uploadUri) {
+ this.leaseClient = Objects.requireNonNull(leaseClient, "leaseClient must not be null");
+ this.lockBlob = Objects.requireNonNull(lockBlob, "lockBlob must not be null");
+ this.uploadUri = Objects.requireNonNull(uploadUri, "uploadUri must not be null");
+
+ // Initialize background daemon thread to renew lease periodically during upload
+ this.renewalExecutor =
+ Utils.scheduleWatchdog(
+ "azure-lease-renewal-" + uploadUri,
+ this::renewLease,
+ RENEWAL_INTERVAL_SECONDS,
+ RENEWAL_INTERVAL_SECONDS,
+ TimeUnit.SECONDS);
+ }
+
+ /** Attempts to renew the lease with Azure Blob Storage. */
+ void renewLease() {
+ if (released) {
+ return;
+ }
+ try {
+ leaseClient.renewLease();
+ log.trace("Successfully renewed Azure blob lease for upload URI {}", uploadUri);
+ } catch (Exception e) {
+ log.warn("Failed to renew Azure blob lease for upload URI {}: {}", uploadUri, e.getMessage());
+ // ponytail: lease was broken externally or expired, shutdown executor
+ released = true;
+ shutdownExecutor();
+ }
+ }
+
+ @Override
+ public void release() {
+ if (!released) {
+ released = true;
+ shutdownExecutor();
+ try {
+ leaseClient.releaseLease();
+ log.trace("Released Azure blob lease for upload URI {}", uploadUri);
+ } catch (Exception e) {
+ log.debug(
+ "Azure blob lease release failed (may have already expired/broken) for URI {}: {}",
+ uploadUri,
+ e.getMessage());
+ }
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ release();
+ }
+
+ @Override
+ public String getUploadUri() {
+ return uploadUri;
+ }
+
+ /** Shuts down the renewal executor cleanly. */
+ private void shutdownExecutor() {
+ Utils.shutdownExecutor(renewalExecutor);
+ }
+}
diff --git a/src/main/java/me/desair/tus/server/upload/azure/AzureErrorType.java b/src/main/java/me/desair/tus/server/upload/azure/AzureErrorType.java
new file mode 100644
index 0000000..002d0d3
--- /dev/null
+++ b/src/main/java/me/desair/tus/server/upload/azure/AzureErrorType.java
@@ -0,0 +1,33 @@
+package me.desair.tus.server.upload.azure;
+
+import com.azure.storage.blob.models.BlobStorageException;
+
+/** Standardized Azure Blob Storage error types parsed from {@link BlobStorageException}. */
+public enum AzureErrorType {
+ /** Target blob or container does not exist (HTTP 404 / BlobNotFound / ContainerNotFound). */
+ BLOB_NOT_FOUND,
+
+ /**
+ * Blob lease is already held by another client / lock contention (HTTP 409 / LeaseAlreadyPresent
+ * / LeaseIdMismatchWithLeaseOperation).
+ */
+ LEASE_ALREADY_PRESENT,
+
+ /** No active lease exists on the blob (HTTP 409 / LeaseNotPresentWithLeaseOperation). */
+ LEASE_NOT_PRESENT,
+
+ /** Conditional request precondition failed (HTTP 412 / ConditionNotMet). */
+ PRECONDITION_FAILED,
+
+ /** Operation or API method not implemented by server/emulator (HTTP 501 / APINotImplemented). */
+ API_NOT_IMPLEMENTED,
+
+ /** Permission denied or invalid SAS credentials (HTTP 403 / AuthorizationFailure). */
+ ACCESS_DENIED,
+
+ /** Lock or resource already exists / general conflict (HTTP 409 / BlobAlreadyExists). */
+ CONFLICT,
+
+ /** Unknown or unmapped Azure storage exception. */
+ UNKNOWN
+}
diff --git a/src/main/java/me/desair/tus/server/upload/azure/AzureUtils.java b/src/main/java/me/desair/tus/server/upload/azure/AzureUtils.java
new file mode 100644
index 0000000..b7e1a10
--- /dev/null
+++ b/src/main/java/me/desair/tus/server/upload/azure/AzureUtils.java
@@ -0,0 +1,67 @@
+package me.desair.tus.server.upload.azure;
+
+import com.azure.storage.blob.models.BlobStorageException;
+
+/** Utility helper methods for parsing and evaluating Azure Blob Storage error responses. */
+public final class AzureUtils {
+
+ private AzureUtils() {
+ // Utility class
+ }
+
+ /**
+ * Parses a {@link BlobStorageException} into a strongly-typed {@link AzureErrorType}.
+ *
+ * @param exception The Azure BlobStorageException to evaluate
+ * @return The corresponding AzureErrorType enum
+ */
+ public static AzureErrorType parseErrorResponse(BlobStorageException exception) {
+ if (exception == null) {
+ return AzureErrorType.UNKNOWN;
+ }
+
+ int statusCode = exception.getStatusCode();
+ String rawErrorCode = "";
+ try {
+ if (exception.getErrorCode() != null) {
+ rawErrorCode = exception.getErrorCode().toString();
+ }
+ } catch (Exception ignored) {
+ // Defensive fallback if exception lacks initialized HTTP headers
+ }
+
+ String errorCodeStr = rawErrorCode.replaceAll("[^a-zA-Z]", "").toLowerCase();
+
+ if (statusCode == 404
+ || errorCodeStr.contains("blobnotfound")
+ || errorCodeStr.contains("containernotfound")) {
+ return AzureErrorType.BLOB_NOT_FOUND;
+ }
+
+ if (statusCode == 409) {
+ if (errorCodeStr.contains("leasealreadypresent")
+ || errorCodeStr.contains("leaseidmismatchwithleaseoperation")) {
+ return AzureErrorType.LEASE_ALREADY_PRESENT;
+ }
+ if (errorCodeStr.contains("leasenotpresentwithleaseoperation")
+ || errorCodeStr.contains("leaseidmissing")) {
+ return AzureErrorType.LEASE_NOT_PRESENT;
+ }
+ return AzureErrorType.CONFLICT;
+ }
+
+ if (statusCode == 412 || errorCodeStr.contains("conditionnotmet")) {
+ return AzureErrorType.PRECONDITION_FAILED;
+ }
+
+ if (statusCode == 501 || errorCodeStr.contains("apinotimplemented")) {
+ return AzureErrorType.API_NOT_IMPLEMENTED;
+ }
+
+ if (statusCode == 403 || errorCodeStr.contains("authorizationfailure")) {
+ return AzureErrorType.ACCESS_DENIED;
+ }
+
+ return AzureErrorType.UNKNOWN;
+ }
+}
diff --git a/src/main/java/me/desair/tus/server/upload/cache/ThreadLocalCachedStorageAndLockingService.java b/src/main/java/me/desair/tus/server/upload/cache/ThreadLocalCachedStorageAndLockingService.java
index 30de697..790c68f 100644
--- a/src/main/java/me/desair/tus/server/upload/cache/ThreadLocalCachedStorageAndLockingService.java
+++ b/src/main/java/me/desair/tus/server/upload/cache/ThreadLocalCachedStorageAndLockingService.java
@@ -260,6 +260,13 @@ public void requestLockRelease(String requestUri) {
lockingServiceDelegate.requestLockRelease(requestUri);
}
+ @Override
+ public void close() throws IOException {
+ lockingServiceDelegate.close();
+ storageServiceDelegate.close();
+ cleanupCache();
+ }
+
private void cleanupCache() {
WeakReference ref = uploadInfoCache.get();
if (ref != null) {
diff --git a/src/main/java/me/desair/tus/server/upload/disk/DiskLockingService.java b/src/main/java/me/desair/tus/server/upload/disk/DiskLockingService.java
index f702c1a..4bbf6d4 100644
--- a/src/main/java/me/desair/tus/server/upload/disk/DiskLockingService.java
+++ b/src/main/java/me/desair/tus/server/upload/disk/DiskLockingService.java
@@ -1,5 +1,6 @@
package me.desair.tus.server.upload.disk;
+import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
@@ -18,6 +19,7 @@
import me.desair.tus.server.upload.UploadLock;
import me.desair.tus.server.upload.UploadLockingService;
import me.desair.tus.server.util.InterruptibleInputStream;
+import me.desair.tus.server.util.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,7 +31,8 @@
* 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 {
+public class DiskLockingService extends AbstractDiskBasedService
+ implements UploadLockingService, Closeable {
private static final Logger log = LoggerFactory.getLogger(DiskLockingService.class);
private static final String LOCK_SUB_DIRECTORY = "locks";
@@ -41,10 +44,15 @@ public class DiskLockingService extends AbstractDiskBasedService implements Uplo
private static Thread watchdogThread = null;
private static final Object watchdogLock = new Object();
+ private final Thread shutdownHook;
+ private volatile boolean closed = false;
+
private UploadIdFactory idFactory;
public DiskLockingService(String storagePath) {
super(storagePath + File.separator + LOCK_SUB_DIRECTORY);
+ this.shutdownHook = new Thread(this::closeQuietly, "disk-lock-shutdown-hook");
+ registerShutdownHook();
}
/** Constructor to use custom UploadIdFactory. */
@@ -54,6 +62,44 @@ public DiskLockingService(UploadIdFactory idFactory, String storagePath) {
this.idFactory = idFactory;
}
+ private void registerShutdownHook() {
+ try {
+ Runtime.getRuntime().addShutdownHook(shutdownHook);
+ } catch (IllegalStateException ignored) {
+ // JVM is already shutting down
+ }
+ }
+
+ private void deregisterShutdownHook() {
+ try {
+ Runtime.getRuntime().removeShutdownHook(shutdownHook);
+ } catch (IllegalStateException ignored) {
+ // JVM is already shutting down
+ }
+ }
+
+ private void closeQuietly() {
+ try {
+ close();
+ } catch (Exception ignored) {
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ synchronized (watchdogLock) {
+ if (watchdogThread != null) {
+ watchdogThread.interrupt();
+ watchdogThread = null;
+ }
+ }
+ activeLocks.clear();
+ if (!closed) {
+ closed = true;
+ deregisterShutdownHook();
+ }
+ }
+
/**
* Attempts to lock the upload resource. Wraps the lock in a RegisteredLock decorator to manage
* cleanup of stop files and the active lock registry.
@@ -161,11 +207,14 @@ public void requestLockRelease(String requestUri) {
// 1. Release JVM-local lock if active
WeakReference streamRef = activeLocks.get(idStr);
if (streamRef != null) {
- InterruptibleInputStream stream = streamRef.get();
- if (stream != null) {
- stream.interrupt();
+ try {
+ InterruptibleInputStream stream = streamRef.get();
+ if (stream != null) {
+ Utils.interruptStream(stream);
+ }
+ } finally {
+ activeLocks.remove(idStr);
}
- activeLocks.remove(idStr);
}
// 2. Create the stop file to signal other replicas
@@ -245,31 +294,32 @@ public void run() {
}
private void checkActiveLocks() {
- // Check stop files for each active lock
+ // Check stop files for each active lock safely per entry
for (Map.Entry> entry :
activeLocks.entrySet()) {
String idStr = entry.getKey();
- WeakReference ref = entry.getValue();
- InterruptibleInputStream stream = ref.get();
+ try {
+ WeakReference ref = entry.getValue();
+ InterruptibleInputStream stream = ref != null ? ref.get() : null;
- if (stream == null) {
+ if (stream == null) {
+ activeLocks.remove(idStr);
+ continue;
+ }
+
+ checkStopFileAndInterrupt(idStr, stream);
+ } catch (Throwable t) {
+ log.warn("Error checking active lock for ID " + idStr, t);
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);
- }
+ log.info("Watchdog detected stop file for upload ID {}. Interrupting stream.", idStr);
+ Utils.interruptStream(stream);
activeLocks.remove(idStr);
try {
Files.deleteIfExists(stopFilePath);
diff --git a/src/main/java/me/desair/tus/server/upload/s3/S3ConcatenationService.java b/src/main/java/me/desair/tus/server/upload/s3/S3ConcatenationService.java
index 9d170b6..f798de5 100644
--- a/src/main/java/me/desair/tus/server/upload/s3/S3ConcatenationService.java
+++ b/src/main/java/me/desair/tus/server/upload/s3/S3ConcatenationService.java
@@ -169,21 +169,24 @@ public void merge(UploadInfo uploadInfo) throws IOException, UploadNotFoundExcep
@Override
public InputStream getConcatenatedBytes(UploadInfo uploadInfo)
throws IOException, UploadNotFoundException {
-
if (uploadInfo == null) {
return null;
}
- if (uploadInfo.getStorageUploadId() == null) {
+ if (uploadStorageService == null) {
+ throw new IOException(
+ "UploadStorageService must be configured to retrieve concatenated upload bytes");
+ }
+
+ if (uploadInfo.isUploadInProgress()) {
merge(uploadInfo);
}
- if (uploadStorageService != null) {
+ if (!uploadInfo.isUploadInProgress()) {
return uploadStorageService.getUploadedBytes(uploadInfo.getId());
}
- throw new IOException(
- "UploadStorageService must be configured to retrieve concatenated upload bytes");
+ return new java.io.ByteArrayInputStream(new byte[0]);
}
@Override
diff --git a/src/main/java/me/desair/tus/server/upload/s3/S3ErrorType.java b/src/main/java/me/desair/tus/server/upload/s3/S3ErrorType.java
index 11b82f7..901b903 100644
--- a/src/main/java/me/desair/tus/server/upload/s3/S3ErrorType.java
+++ b/src/main/java/me/desair/tus/server/upload/s3/S3ErrorType.java
@@ -16,6 +16,9 @@ public enum S3ErrorType {
/** Permission denied or invalid credentials (HTTP 403 / AccessDenied). */
ACCESS_DENIED,
+ /** Operation or API method not implemented by server/emulator (HTTP 501 / APINotImplemented). */
+ API_NOT_IMPLEMENTED,
+
/** Unknown or unmapped S3 error response. */
UNKNOWN
}
diff --git a/src/main/java/me/desair/tus/server/upload/s3/S3LockingService.java b/src/main/java/me/desair/tus/server/upload/s3/S3LockingService.java
index 823f8db..3989230 100644
--- a/src/main/java/me/desair/tus/server/upload/s3/S3LockingService.java
+++ b/src/main/java/me/desair/tus/server/upload/s3/S3LockingService.java
@@ -10,13 +10,13 @@
import io.minio.errors.ErrorResponseException;
import io.minio.messages.Item;
import java.io.ByteArrayInputStream;
+import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import me.desair.tus.server.exception.TusException;
@@ -26,8 +26,8 @@
import me.desair.tus.server.upload.UploadLock;
import me.desair.tus.server.upload.UploadLockingService;
import me.desair.tus.server.upload.UuidUploadIdFactory;
-import me.desair.tus.server.util.InterruptibleInputStream;
import me.desair.tus.server.util.S3UploadLockJsonSerializer;
+import me.desair.tus.server.util.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,7 +54,7 @@
* pods.
*
*/
-public class S3LockingService implements UploadLockingService {
+public class S3LockingService implements UploadLockingService, Closeable {
private static final Logger log = LoggerFactory.getLogger(S3LockingService.class);
@@ -72,6 +72,9 @@ public class S3LockingService implements UploadLockingService {
private final Map activeInputStreams = new ConcurrentHashMap<>();
private final ScheduledExecutorService watchdogExecutor;
+ private final Thread shutdownHook;
+ private volatile boolean closed = false;
+
/**
* Basic constructor using default lock prefix ("locks/"), 30s lease duration, and 2s polling
* interval.
@@ -111,16 +114,38 @@ public S3LockingService(
// Background watchdog thread to poll S3 for .stop contention signals across pods
this.watchdogExecutor =
- Executors.newSingleThreadScheduledExecutor(
- r -> {
- Thread t = new Thread(r, "s3-lock-watchdog");
- t.setDaemon(true);
- return t;
- });
-
- if (pollIntervalMs > 0) {
- this.watchdogExecutor.scheduleAtFixedRate(
- this::checkStopSignals, pollIntervalMs, pollIntervalMs, TimeUnit.MILLISECONDS);
+ Utils.scheduleWatchdog(
+ "s3-lock-watchdog",
+ this::checkStopSignals,
+ pollIntervalMs,
+ pollIntervalMs,
+ TimeUnit.MILLISECONDS);
+
+ // Register automatic JVM shutdown hook to clean up watchdog thread pool on app/pod shutdown
+ this.shutdownHook = new Thread(this::closeQuietly, "s3-lock-shutdown-hook");
+ registerShutdownHook();
+ }
+
+ private void registerShutdownHook() {
+ try {
+ Runtime.getRuntime().addShutdownHook(shutdownHook);
+ } catch (IllegalStateException ignored) {
+ // JVM is already shutting down
+ }
+ }
+
+ private void deregisterShutdownHook() {
+ try {
+ Runtime.getRuntime().removeShutdownHook(shutdownHook);
+ } catch (IllegalStateException ignored) {
+ // JVM is already shutting down
+ }
+ }
+
+ private void closeQuietly() {
+ try {
+ close();
+ } catch (Exception ignored) {
}
}
@@ -203,9 +228,9 @@ public void requestLockRelease(String requestUri) {
}
// Step 1: Interrupt local active payload byte stream if hosted on this node
- InputStream activeStream = activeInputStreams.get(requestUri);
+ InputStream activeStream = activeInputStreams.remove(requestUri);
if (activeStream != null) {
- interruptStream(activeStream);
+ Utils.interruptStream(activeStream);
}
// Step 2: Write a .stop signal object to S3 to signal lock contention across remote nodes/pods
@@ -286,6 +311,16 @@ private void writeStopSignal(UploadId uploadId) {
}
}
+ @Override
+ public void close() throws IOException {
+ if (!closed) {
+ closed = true;
+ deregisterShutdownHook();
+ Utils.shutdownExecutor(watchdogExecutor);
+ activeInputStreams.clear();
+ }
+ }
+
private void checkStopSignals() {
for (Map.Entry entry : activeInputStreams.entrySet()) {
checkStopSignalForEntry(entry.getKey(), entry.getValue());
@@ -305,7 +340,7 @@ private void checkStopSignalForEntry(String uri, InputStream inputStream) {
// Remote stop signal object found! Interrupt local byte stream immediately
interruptStream(inputStream);
} catch (ErrorResponseException e) {
- if ("NoSuchKey".equalsIgnoreCase(e.errorResponse().code())) {
+ if (S3Utils.parseErrorResponse(e) == S3ErrorType.NO_SUCH_KEY) {
// Normal state: no stop signal object in S3
return;
}
@@ -315,14 +350,7 @@ private void checkStopSignalForEntry(String uri, InputStream inputStream) {
}
private void interruptStream(InputStream is) {
- if (is instanceof InterruptibleInputStream) {
- ((InterruptibleInputStream) is).interrupt();
- } else {
- try {
- is.close();
- } catch (Exception ignored) {
- }
- }
+ Utils.interruptStream(is);
}
private void deleteObjectQuietly(String key) {
diff --git a/src/main/java/me/desair/tus/server/upload/s3/S3StorageService.java b/src/main/java/me/desair/tus/server/upload/s3/S3StorageService.java
index 5a41b88..8045d02 100644
--- a/src/main/java/me/desair/tus/server/upload/s3/S3StorageService.java
+++ b/src/main/java/me/desair/tus/server/upload/s3/S3StorageService.java
@@ -172,6 +172,11 @@ public String getS3ObjectKey(UploadInfo uploadInfo) {
return null;
}
+ // Resolve duplicate child upload dynamically to parent S3 object key per AGENTS.md §7
+ if (uploadInfo.getDuplicatesUploadId() != null) {
+ return buildObjectKey(uploadInfo.getDuplicatesUploadId());
+ }
+
if (uploadInfo.getStorageUploadId() != null) {
return uploadInfo.getStorageUploadId();
} else {
@@ -372,7 +377,7 @@ public InputStream getUploadedBytes(UploadId id) throws IOException, UploadNotFo
}
// Handle concatenated upload resolution if applicable
- if (UploadType.CONCATENATED.equals(info.getUploadType()) && info.getStorageUploadId() == null) {
+ if (UploadType.CONCATENATED.equals(info.getUploadType()) && info.isUploadInProgress()) {
if (concatenationService != null) {
concatenationService.merge(info);
info = getUploadInfo(id);
@@ -466,6 +471,10 @@ public void terminateUpload(UploadInfo uploadInfo) throws UploadNotFoundExceptio
deleteObjectQuietly(
buildChecksumKey(uploadInfo.getChecksum(), uploadInfo.getChecksumAlgorithm()));
}
+
+ // Delete lock target and stop signal objects
+ deleteObjectQuietly(buildLockKey(uploadInfo.getId()));
+ deleteObjectQuietly(buildStopKey(uploadInfo.getId()));
}
@Override
@@ -630,7 +639,7 @@ private InputStream prepareStreamWithExistingIncompletePart(
return new SequenceInputStream(new FileInputStream(tempPrependedFile), inputStream);
}
} catch (ErrorResponseException e) {
- if ("NoSuchKey".equalsIgnoreCase(e.errorResponse().code())) {
+ if (S3Utils.parseErrorResponse(e) == S3ErrorType.NO_SUCH_KEY) {
// Normal case: no leftover .part object present in S3
}
} catch (Exception ignored) {
@@ -1016,6 +1025,14 @@ private String buildChecksumKey(String checksum, ChecksumAlgorithm algorithm) {
return checksumsPrefix + algorithmName + "/" + checksum;
}
+ private String buildLockKey(UploadId id) {
+ return locksPrefix + id.toString() + ".lock";
+ }
+
+ private String buildStopKey(UploadId id) {
+ return locksPrefix + id.toString() + ".stop";
+ }
+
private static class AppendResult {
final long totalBytesAppended;
final List allPartKeys;
diff --git a/src/main/java/me/desair/tus/server/upload/s3/S3UploadLock.java b/src/main/java/me/desair/tus/server/upload/s3/S3UploadLock.java
index 4a4303d..91d43a7 100644
--- a/src/main/java/me/desair/tus/server/upload/s3/S3UploadLock.java
+++ b/src/main/java/me/desair/tus/server/upload/s3/S3UploadLock.java
@@ -9,11 +9,11 @@
import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.util.Map;
-import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import me.desair.tus.server.upload.UploadLock;
import me.desair.tus.server.util.S3UploadLockJsonSerializer;
+import me.desair.tus.server.util.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -129,14 +129,12 @@ public S3UploadLock(
// lease)
long heartbeatPeriodMs = Math.max(1000L, leaseDurationMs / 3);
this.heartbeatExecutor =
- Executors.newSingleThreadScheduledExecutor(
- r -> {
- Thread t = new Thread(r, "s3-lock-heartbeat-" + holderId);
- t.setDaemon(true);
- return t;
- });
- this.heartbeatExecutor.scheduleAtFixedRate(
- this::renewLease, heartbeatPeriodMs, heartbeatPeriodMs, TimeUnit.MILLISECONDS);
+ Utils.scheduleWatchdog(
+ "s3-lock-heartbeat-" + holderId,
+ this::renewLease,
+ heartbeatPeriodMs,
+ heartbeatPeriodMs,
+ TimeUnit.MILLISECONDS);
}
S3UploadLock(
@@ -239,13 +237,7 @@ public void release() {
@Override
public void close() {
// Step 1: Stop the background heartbeat daemon thread
- if (heartbeatExecutor != null) {
- try {
- heartbeatExecutor.shutdownNow();
- } catch (Exception e) {
- log.debug("Error shutting down lock heartbeat executor", e);
- }
- }
+ Utils.shutdownExecutor(heartbeatExecutor);
// Step 2: Remove active request stream registration
if (inputStreamMap != null && requestUri != null) {
diff --git a/src/main/java/me/desair/tus/server/upload/s3/S3Utils.java b/src/main/java/me/desair/tus/server/upload/s3/S3Utils.java
index 675971e..618f41c 100644
--- a/src/main/java/me/desair/tus/server/upload/s3/S3Utils.java
+++ b/src/main/java/me/desair/tus/server/upload/s3/S3Utils.java
@@ -22,9 +22,11 @@ public static S3ErrorType parseErrorResponse(ErrorResponseException exception) {
}
ErrorResponse response = exception.errorResponse();
- String code = response != null ? response.code() : "";
+ String code = response != null && response.code() != null ? response.code() : "";
- if ("NoSuchKey".equalsIgnoreCase(code) || "NoSuchBucket".equalsIgnoreCase(code)) {
+ if ("NoSuchKey".equalsIgnoreCase(code)
+ || "NoSuchBucket".equalsIgnoreCase(code)
+ || "NoSuchUpload".equalsIgnoreCase(code)) {
return S3ErrorType.NO_SUCH_KEY;
}
@@ -41,6 +43,10 @@ public static S3ErrorType parseErrorResponse(ErrorResponseException exception) {
return S3ErrorType.ACCESS_DENIED;
}
+ if ("APINotImplemented".equalsIgnoreCase(code) || "NotImplemented".equalsIgnoreCase(code)) {
+ return S3ErrorType.API_NOT_IMPLEMENTED;
+ }
+
return S3ErrorType.UNKNOWN;
}
}
diff --git a/src/main/java/me/desair/tus/server/util/Utils.java b/src/main/java/me/desair/tus/server/util/Utils.java
index 94da342..776fbd6 100644
--- a/src/main/java/me/desair/tus/server/util/Utils.java
+++ b/src/main/java/me/desair/tus/server/util/Utils.java
@@ -19,6 +19,9 @@
import java.util.EnumSet;
import java.util.LinkedList;
import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import me.desair.tus.server.HttpHeader;
import me.desair.tus.server.HttpMethod;
@@ -389,4 +392,77 @@ public static String getUploadUriOnCreation(
uploadInfo != null && uploadInfo.getId() != null ? uploadInfo.getId().toString() : "";
return baseUri + (baseUri.endsWith("/") ? "" : "/") + idStr;
}
+
+ /**
+ * Creates a single-thread scheduled executor service with a daemon thread of the given name.
+ *
+ * @param threadName The name for the background daemon thread
+ * @return A new single-thread ScheduledExecutorService
+ */
+ public static ScheduledExecutorService createScheduledDaemonExecutor(String threadName) {
+ return Executors.newSingleThreadScheduledExecutor(
+ runnable -> {
+ Thread thread = new Thread(runnable, threadName);
+ thread.setDaemon(true);
+ return thread;
+ });
+ }
+
+ /**
+ * Creates a daemon scheduled executor and immediately schedules a task to run periodically at a
+ * fixed rate.
+ *
+ * @param threadName The name for the background daemon thread
+ * @param task The task to execute periodically
+ * @param initialDelay The initial delay before the first execution
+ * @param period The period between successive executions
+ * @param unit The time unit of the initialDelay and period parameters
+ * @return The created ScheduledExecutorService
+ */
+ public static ScheduledExecutorService scheduleWatchdog(
+ String threadName, Runnable task, long initialDelay, long period, TimeUnit unit) {
+ ScheduledExecutorService executor = createScheduledDaemonExecutor(threadName);
+ if (period > 0 && task != null) {
+ executor.scheduleAtFixedRate(task, initialDelay, period, unit);
+ }
+ return executor;
+ }
+
+ /**
+ * Safely shuts down a ScheduledExecutorService using {@link
+ * ScheduledExecutorService#shutdownNow()}.
+ *
+ * @param executor The ScheduledExecutorService to shut down
+ */
+ public static void shutdownExecutor(ScheduledExecutorService executor) {
+ if (executor != null && !executor.isShutdown()) {
+ try {
+ executor.shutdownNow();
+ } catch (Exception e) {
+ log.debug("Error shutting down executor: {}", e.getMessage());
+ }
+ }
+ }
+
+ /**
+ * Safely interrupts an input stream if it is an instance of {@link InterruptibleInputStream}, or
+ * closes it quietly if it is a standard input stream. Any exceptions encountered during
+ * interruption or closing are caught and logged without propagating.
+ *
+ * @param inputStream The InputStream to interrupt or close
+ */
+ public static void interruptStream(java.io.InputStream inputStream) {
+ if (inputStream == null) {
+ return;
+ }
+ try {
+ if (inputStream instanceof InterruptibleInputStream) {
+ ((InterruptibleInputStream) inputStream).interrupt();
+ } else {
+ inputStream.close();
+ }
+ } catch (Throwable t) {
+ log.warn("Error interrupting input stream: {}", t.getMessage(), t);
+ }
+ }
}
diff --git a/src/test/java/me/desair/tus/server/AbstractITTusFileUploadService.java b/src/test/java/me/desair/tus/server/AbstractITTusFileUploadService.java
index c558fb3..77afd56 100644
--- a/src/test/java/me/desair/tus/server/AbstractITTusFileUploadService.java
+++ b/src/test/java/me/desair/tus/server/AbstractITTusFileUploadService.java
@@ -1054,6 +1054,105 @@ public void testConcatenationCompleted() throws Exception {
}
}
+ @Test
+ public void testConcatenationOutOfOrderCompleted() throws Exception {
+ String part1 = "This is the first part of my test upload ";
+ String part2 = "and this is the second part.";
+
+ // 1. Create the SECOND upload part FIRST
+ servletRequest.setMethod("POST");
+ servletRequest.setRequestURI(UPLOAD_URI);
+ servletRequest.addHeader(HttpHeader.CONTENT_LENGTH, 0);
+ servletRequest.addHeader(HttpHeader.UPLOAD_LENGTH, "28");
+ servletRequest.addHeader(HttpHeader.TUS_RESUMABLE, "1.0.0");
+ servletRequest.addHeader(HttpHeader.UPLOAD_CONCAT, "partial");
+ servletRequest.addHeader(HttpHeader.UPLOAD_METADATA, "filename cGFydDIudHh0");
+
+ tusFileUploadService.process(servletRequest, servletResponse, OWNER_KEY);
+ assertResponseStatus(HttpServletResponse.SC_CREATED);
+ String location2 =
+ UPLOAD_URI
+ + StringUtils.substringAfter(
+ servletResponse.getHeader(HttpHeader.LOCATION), UPLOAD_URI);
+
+ // Upload part 2 bytes first
+ reset();
+ servletRequest.setMethod("PATCH");
+ servletRequest.setRequestURI(location2);
+ servletRequest.addHeader(HttpHeader.CONTENT_TYPE, "application/offset+octet-stream");
+ servletRequest.addHeader(HttpHeader.CONTENT_LENGTH, "28");
+ servletRequest.addHeader(HttpHeader.UPLOAD_OFFSET, "0");
+ servletRequest.addHeader(HttpHeader.TUS_RESUMABLE, "1.0.0");
+ servletRequest.setContent(part2.getBytes());
+
+ tusFileUploadService.process(servletRequest, servletResponse, OWNER_KEY);
+ assertResponseStatus(HttpServletResponse.SC_NO_CONTENT);
+
+ // 2. Create the FIRST upload part SECOND
+ reset();
+ servletRequest.setMethod("POST");
+ servletRequest.setRequestURI(UPLOAD_URI);
+ servletRequest.addHeader(HttpHeader.CONTENT_LENGTH, 0);
+ servletRequest.addHeader(HttpHeader.UPLOAD_LENGTH, "41");
+ servletRequest.addHeader(HttpHeader.TUS_RESUMABLE, "1.0.0");
+ servletRequest.addHeader(HttpHeader.UPLOAD_CONCAT, "partial");
+ servletRequest.addHeader(HttpHeader.UPLOAD_METADATA, "filename cGFydDEudHh0");
+
+ tusFileUploadService.process(servletRequest, servletResponse, OWNER_KEY);
+ assertResponseStatus(HttpServletResponse.SC_CREATED);
+ String location1 =
+ UPLOAD_URI
+ + StringUtils.substringAfter(
+ servletResponse.getHeader(HttpHeader.LOCATION), UPLOAD_URI);
+
+ // Upload part 1 bytes second
+ reset();
+ servletRequest.setMethod("PATCH");
+ servletRequest.setRequestURI(location1);
+ servletRequest.addHeader(HttpHeader.CONTENT_TYPE, "application/offset+octet-stream");
+ servletRequest.addHeader(HttpHeader.CONTENT_LENGTH, "41");
+ servletRequest.addHeader(HttpHeader.UPLOAD_OFFSET, "0");
+ servletRequest.addHeader(HttpHeader.TUS_RESUMABLE, "1.0.0");
+ servletRequest.setContent(part1.getBytes());
+
+ tusFileUploadService.process(servletRequest, servletResponse, OWNER_KEY);
+ assertResponseStatus(HttpServletResponse.SC_NO_CONTENT);
+
+ // 3. Create final concatenated upload referencing part 1 then part 2 in correct order
+ reset();
+ servletRequest.setMethod("POST");
+ servletRequest.setRequestURI(UPLOAD_URI);
+ servletRequest.addHeader(HttpHeader.CONTENT_LENGTH, 0);
+ servletRequest.addHeader(HttpHeader.TUS_RESUMABLE, "1.0.0");
+ servletRequest.addHeader(HttpHeader.UPLOAD_CONCAT, "final ; " + location1 + " " + location2);
+
+ tusFileUploadService.process(servletRequest, servletResponse, OWNER_KEY);
+ assertResponseStatus(HttpServletResponse.SC_CREATED);
+ String finalLocation =
+ UPLOAD_URI
+ + StringUtils.substringAfter(
+ servletResponse.getHeader(HttpHeader.LOCATION), UPLOAD_URI);
+
+ // Download and verify content is in the correct concatenated order (part1 + part2)
+ reset();
+ servletRequest.setMethod("GET");
+ servletRequest.setRequestURI(finalLocation);
+ servletRequest.addHeader(HttpHeader.TUS_RESUMABLE, "1.0.0");
+
+ tusFileUploadService.process(servletRequest, servletResponse, OWNER_KEY);
+ assertResponseStatus(HttpServletResponse.SC_OK);
+ assertThat(
+ servletResponse.getContentAsString(),
+ is("This is the first part of my test upload and this is the second part."));
+
+ try (InputStream uploadedBytes =
+ tusFileUploadService.getUploadedBytes(finalLocation, OWNER_KEY)) {
+ assertThat(
+ IOUtils.toString(uploadedBytes, StandardCharsets.UTF_8),
+ is("This is the first part of my test upload and this is the second part."));
+ }
+ }
+
@Test
public void testConcatenationUnfinished() throws Exception {
String part1 = "When sending this part, the final upload was already created. ";
diff --git a/src/test/java/me/desair/tus/server/TestUtils.java b/src/test/java/me/desair/tus/server/TestUtils.java
index 7a48b21..3d698f7 100644
--- a/src/test/java/me/desair/tus/server/TestUtils.java
+++ b/src/test/java/me/desair/tus/server/TestUtils.java
@@ -104,4 +104,47 @@ public static void createBucket(MinioClient minioClient, String bucket) {
throw new RuntimeException("Failed to create bucket " + bucket, e);
}
}
+
+ /**
+ * Create and configure a GenericContainer running Azurite for Azure integration testing.
+ *
+ * @return A configured GenericContainer instance (not started yet)
+ */
+ public static GenericContainer createAzuriteContainer() {
+ return new GenericContainer<>("mcr.microsoft.com/azure-storage/azurite:3.36.0")
+ .withExposedPorts(10000)
+ .withCommand("azurite-blob", "--blobHost", "0.0.0.0", "--skipApiVersionCheck");
+ }
+
+ /**
+ * Create a {@link com.azure.storage.blob.BlobContainerClient} connected to Azurite Testcontainer.
+ *
+ * @param azurite The active Azurite Testcontainer
+ * @param containerName Target container name
+ * @return Pre-configured BlobContainerClient
+ */
+ public static com.azure.storage.blob.BlobContainerClient createBlobContainerClient(
+ GenericContainer azurite, String containerName) {
+ String connectionString =
+ String.format(
+ "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;"
+ + "AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/"
+ + "K1SZFPTOtr/KBHBeksoGMGw==;"
+ + "BlobEndpoint=http://%s:%d/devstoreaccount1",
+ azurite.getHost(), azurite.getMappedPort(10000));
+
+ com.azure.storage.blob.BlobServiceClient serviceClient =
+ new com.azure.storage.blob.BlobServiceClientBuilder()
+ .connectionString(connectionString)
+ .buildClient();
+
+ com.azure.storage.blob.BlobContainerClient containerClient =
+ serviceClient.getBlobContainerClient(containerName);
+
+ if (!containerClient.exists()) {
+ containerClient.create();
+ }
+
+ return containerClient;
+ }
}
diff --git a/src/test/java/me/desair/tus/server/TusFileUploadServiceTest.java b/src/test/java/me/desair/tus/server/TusFileUploadServiceTest.java
index 55a2eee..a1e9210 100644
--- a/src/test/java/me/desair/tus/server/TusFileUploadServiceTest.java
+++ b/src/test/java/me/desair/tus/server/TusFileUploadServiceTest.java
@@ -522,4 +522,54 @@ public void testWithJsonSerialization() {
service.withJsonSerialization(false);
org.junit.Assert.assertFalse(service.getUploadStorageService().isJsonSerializationEnabled());
}
+
+ @Test
+ public void testClose() throws Exception {
+ UploadLockingService mockLockingService = mock(UploadLockingService.class);
+ TusFileUploadService service =
+ new TusFileUploadService().withUploadLockingService(mockLockingService);
+
+ service.close();
+
+ verify(mockLockingService).close();
+ }
+
+ @Test
+ public void testGetUploadInfoSingleArg() throws Exception {
+ UploadLockingService mockLockingService = mock(UploadLockingService.class);
+ UploadStorageService mockStorageService = mock(UploadStorageService.class);
+ UploadLock mockLock = mock(UploadLock.class);
+ UploadInfo mockInfo = new UploadInfo();
+
+ when(mockLockingService.lockUploadByUri(anyString())).thenReturn(mockLock);
+ when(mockStorageService.getUploadInfo("/files/123", null)).thenReturn(mockInfo);
+
+ TusFileUploadService service =
+ new TusFileUploadService()
+ .withUploadLockingService(mockLockingService)
+ .withUploadStorageService(mockStorageService);
+
+ UploadInfo result = service.getUploadInfo("/files/123");
+ assertNotNull(result);
+ verify(mockStorageService).getUploadInfo("/files/123", null);
+ }
+
+ @Test
+ public void testDeleteUploadSingleArg() throws Exception {
+ UploadLockingService mockLockingService = mock(UploadLockingService.class);
+ UploadStorageService mockStorageService = mock(UploadStorageService.class);
+ UploadLock mockLock = mock(UploadLock.class);
+ UploadInfo mockInfo = new UploadInfo();
+
+ when(mockLockingService.lockUploadByUri(anyString())).thenReturn(mockLock);
+ when(mockStorageService.getUploadInfo("/files/123", null)).thenReturn(mockInfo);
+
+ TusFileUploadService service =
+ new TusFileUploadService()
+ .withUploadLockingService(mockLockingService)
+ .withUploadStorageService(mockStorageService);
+
+ service.deleteUpload("/files/123");
+ verify(mockStorageService).terminateUpload(mockInfo);
+ }
}
diff --git a/src/test/java/me/desair/tus/server/upload/UploadLockingServiceTest.java b/src/test/java/me/desair/tus/server/upload/UploadLockingServiceTest.java
index d63cf59..10b7d43 100644
--- a/src/test/java/me/desair/tus/server/upload/UploadLockingServiceTest.java
+++ b/src/test/java/me/desair/tus/server/upload/UploadLockingServiceTest.java
@@ -1,23 +1,25 @@
package me.desair.tus.server.upload;
-import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertFalse;
import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import me.desair.tus.server.exception.TusException;
import org.junit.Test;
public class UploadLockingServiceTest {
@Test
- public void testDefaultMethods() {
- UploadLockingService service =
+ public void testDefaultMethods() throws Exception {
+ UploadLockingService dummyService =
new UploadLockingService() {
@Override
- public UploadLock lockUploadByUri(String requestUri) {
+ public UploadLock lockUploadByUri(String requestUri) throws TusException, IOException {
return null;
}
@Override
- public void cleanupStaleLocks() {}
+ public void cleanupStaleLocks() throws IOException {}
@Override
public boolean isLocked(UploadId id) {
@@ -28,9 +30,11 @@ public boolean isLocked(UploadId id) {
public void setIdFactory(UploadIdFactory idFactory) {}
};
- // Verify default methods do not throw exceptions and act as no-ops
- service.registerInputStream("/files/test", new ByteArrayInputStream(new byte[0]));
- service.requestLockRelease("/files/test");
- assertNotNull(service);
+ // Test default methods for coverage
+ dummyService.registerInputStream("/test/upload/123", new ByteArrayInputStream(new byte[0]));
+ dummyService.requestLockRelease("/test/upload/123");
+ dummyService.close();
+
+ assertFalse(dummyService.isLocked(new UploadId("123")));
}
}
diff --git a/src/test/java/me/desair/tus/server/upload/azure/AzureUtilsTest.java b/src/test/java/me/desair/tus/server/upload/azure/AzureUtilsTest.java
new file mode 100644
index 0000000..b259c0e
--- /dev/null
+++ b/src/test/java/me/desair/tus/server/upload/azure/AzureUtilsTest.java
@@ -0,0 +1,79 @@
+package me.desair.tus.server.upload.azure;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.azure.core.http.HttpHeaders;
+import com.azure.core.http.HttpResponse;
+import com.azure.storage.blob.models.BlobErrorCode;
+import com.azure.storage.blob.models.BlobStorageException;
+import org.junit.Test;
+
+public class AzureUtilsTest {
+
+ private BlobStorageException createException(int statusCode, BlobErrorCode errorCode) {
+ HttpResponse response = mock(HttpResponse.class);
+ when(response.getStatusCode()).thenReturn(statusCode);
+ HttpHeaders headers = new HttpHeaders();
+ if (errorCode != null) {
+ headers.set("x-ms-error-code", errorCode.toString());
+ }
+ when(response.getHeaders()).thenReturn(headers);
+ return new BlobStorageException("Test exception", response, errorCode);
+ }
+
+ @Test
+ public void testParseErrorResponseNull() {
+ assertEquals(AzureErrorType.UNKNOWN, AzureUtils.parseErrorResponse(null));
+ }
+
+ @Test
+ public void testParseErrorResponseBlobNotFound() {
+ BlobStorageException ex = createException(404, BlobErrorCode.BLOB_NOT_FOUND);
+ assertEquals(AzureErrorType.BLOB_NOT_FOUND, AzureUtils.parseErrorResponse(ex));
+ }
+
+ @Test
+ public void testParseErrorResponseLeaseAlreadyPresent() {
+ BlobStorageException ex = createException(409, BlobErrorCode.LEASE_ALREADY_PRESENT);
+ assertEquals(AzureErrorType.LEASE_ALREADY_PRESENT, AzureUtils.parseErrorResponse(ex));
+ }
+
+ @Test
+ public void testParseErrorResponseLeaseNotPresent() {
+ BlobStorageException ex =
+ createException(409, BlobErrorCode.LEASE_NOT_PRESENT_WITH_LEASE_OPERATION);
+ assertEquals(AzureErrorType.LEASE_NOT_PRESENT, AzureUtils.parseErrorResponse(ex));
+ }
+
+ @Test
+ public void testParseErrorResponseConflict() {
+ BlobStorageException ex = createException(409, BlobErrorCode.BLOB_ALREADY_EXISTS);
+ assertEquals(AzureErrorType.CONFLICT, AzureUtils.parseErrorResponse(ex));
+ }
+
+ @Test
+ public void testParseErrorResponsePreconditionFailed() {
+ BlobStorageException ex = createException(412, BlobErrorCode.CONDITION_NOT_MET);
+ assertEquals(AzureErrorType.PRECONDITION_FAILED, AzureUtils.parseErrorResponse(ex));
+ }
+
+ @Test
+ public void testParseErrorResponseApiNotImplemented() {
+ BlobStorageException ex = createException(501, null);
+ assertEquals(AzureErrorType.API_NOT_IMPLEMENTED, AzureUtils.parseErrorResponse(ex));
+ }
+
+ @Test
+ public void testParseErrorResponseAccessDenied() {
+ BlobStorageException ex = createException(403, BlobErrorCode.AUTHORIZATION_FAILURE);
+ assertEquals(AzureErrorType.ACCESS_DENIED, AzureUtils.parseErrorResponse(ex));
+ }
+
+ @Test
+ public void testParseErrorResponseUnknown() {
+ BlobStorageException ex = createException(500, null);
+ assertEquals(AzureErrorType.UNKNOWN, AzureUtils.parseErrorResponse(ex));
+ }
+}
diff --git a/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobConcatenationService.java b/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobConcatenationService.java
new file mode 100644
index 0000000..c774db7
--- /dev/null
+++ b/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobConcatenationService.java
@@ -0,0 +1,272 @@
+package me.desair.tus.server.upload.azure;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import com.azure.storage.blob.BlobContainerClient;
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import me.desair.tus.server.TestUtils;
+import me.desair.tus.server.exception.UploadNotFoundException;
+import me.desair.tus.server.upload.UploadInfo;
+import me.desair.tus.server.upload.UploadType;
+import org.junit.AfterClass;
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.testcontainers.containers.GenericContainer;
+
+public class ITAzureBlobConcatenationService {
+
+ private static GenericContainer azuriteContainer;
+
+ @BeforeClass
+ public static void setUpClass() {
+ Assume.assumeTrue(TestUtils.isContainerRuntimeAvailable());
+ azuriteContainer = TestUtils.createAzuriteContainer();
+ azuriteContainer.start();
+ }
+
+ @AfterClass
+ public static void tearDownClass() {
+ if (azuriteContainer != null) {
+ azuriteContainer.stop();
+ }
+ }
+
+ private BlobContainerClient containerClient;
+ private AzureBlobStorageService storageService;
+ private AzureBlobConcatenationService concatenationService;
+
+ @Before
+ public void setUp() {
+ Assume.assumeTrue(TestUtils.isContainerRuntimeAvailable() && azuriteContainer != null);
+ containerClient =
+ TestUtils.createBlobContainerClient(
+ azuriteContainer, "concat-unit-container-" + System.nanoTime());
+ storageService = new AzureBlobStorageService(containerClient);
+ concatenationService = new AzureBlobConcatenationService(containerClient, storageService);
+ storageService.setUploadConcatenationService(concatenationService);
+ }
+
+ @Test
+ public void mergeShouldDoNothingIfInfoIsNull() throws Exception {
+ concatenationService.merge(null);
+ }
+
+ @Test
+ public void mergeShouldDoNothingIfConcatFilesIsNull() throws Exception {
+ UploadInfo info = new UploadInfo();
+ concatenationService.merge(info);
+ }
+
+ @Test(expected = UploadNotFoundException.class)
+ public void mergeShouldThrowWhenPartialUploadNotFound() throws Exception {
+ UploadInfo finalInfo = new UploadInfo();
+ finalInfo.setId(new me.desair.tus.server.upload.UploadId("final-id"));
+ finalInfo.setConcatenationPartIds(Arrays.asList("/test/upload/non-existing"));
+ finalInfo.setUploadType(UploadType.CONCATENATED);
+
+ concatenationService.merge(finalInfo);
+ }
+
+ @Test
+ public void mergeShouldStageBlocksAndCommitBlockList() throws Exception {
+ UploadInfo part1Info = new UploadInfo();
+ part1Info.setLength(10L);
+ UploadInfo part1 = storageService.create(part1Info, null);
+ storageService.append(part1, new ByteArrayInputStream("part1-data".getBytes()));
+
+ UploadInfo part2Info = new UploadInfo();
+ part2Info.setLength(10L);
+ UploadInfo part2 = storageService.create(part2Info, null);
+ storageService.append(part2, new ByteArrayInputStream("part2-data".getBytes()));
+
+ UploadInfo finalInfo = new UploadInfo();
+ finalInfo.setConcatenationPartIds(
+ Arrays.asList("/test/upload/" + part1.getId(), "/test/upload/" + part2.getId()));
+ finalInfo.setUploadType(UploadType.CONCATENATED);
+ UploadInfo createdFinal = storageService.create(finalInfo, null);
+
+ concatenationService.merge(createdFinal);
+
+ assertEquals(Long.valueOf(20L), createdFinal.getOffset());
+ assertEquals(Long.valueOf(20L), createdFinal.getLength());
+ }
+
+ @Test
+ public void getConcatenationBytesShouldReturnCombinedStream() throws Exception {
+ UploadInfo part1Info = new UploadInfo();
+ part1Info.setLength(5L);
+ UploadInfo part1 = storageService.create(part1Info, null);
+ storageService.append(part1, new ByteArrayInputStream("hello".getBytes()));
+
+ UploadInfo part2Info = new UploadInfo();
+ part2Info.setLength(6L);
+ UploadInfo part2 = storageService.create(part2Info, null);
+ storageService.append(part2, new ByteArrayInputStream("-world".getBytes()));
+
+ UploadInfo finalInfo = new UploadInfo();
+ finalInfo.setConcatenationPartIds(
+ Arrays.asList("/test/upload/" + part1.getId(), "/test/upload/" + part2.getId()));
+ finalInfo.setUploadType(UploadType.CONCATENATED);
+ UploadInfo createdFinal = storageService.create(finalInfo, null);
+
+ InputStream is = concatenationService.getConcatenatedBytes(createdFinal);
+ assertNotNull(is);
+ assertEquals(
+ "hello-world",
+ org.apache.commons.io.IOUtils.toString(is, java.nio.charset.StandardCharsets.UTF_8));
+ }
+
+ @Test(expected = UploadNotFoundException.class)
+ public void getConcatenatedBytesShouldThrowOnNullInfo() throws Exception {
+ concatenationService.getConcatenatedBytes(null);
+ }
+
+ @Test
+ public void getPartialUploadsShouldReturnList() throws Exception {
+ UploadInfo part1Info = new UploadInfo();
+ part1Info.setLength(5L);
+ UploadInfo part1 = storageService.create(part1Info, null);
+
+ UploadInfo finalInfo = new UploadInfo();
+ finalInfo.setConcatenationPartIds(Arrays.asList("/test/upload/" + part1.getId()));
+
+ List partials = concatenationService.getPartialUploads(finalInfo);
+ assertEquals(1, partials.size());
+ assertEquals(part1.getId(), partials.get(0).getId());
+ }
+
+ @Test
+ public void getPartialUploadsShouldReturnEmptyOnNullInfo() throws Exception {
+ assertTrue(concatenationService.getPartialUploads(null).isEmpty());
+ }
+
+ @Test
+ public void getPartialUploadsShouldReturnEmptyOnNullPartIds() throws Exception {
+ UploadInfo info = new UploadInfo();
+ info.setConcatenationPartIds(null);
+ assertTrue(concatenationService.getPartialUploads(info).isEmpty());
+ }
+
+ @Test
+ public void mergeShouldUpdateExpirationWhenExpirationPeriodIsSet() throws Exception {
+ storageService.setUploadExpirationPeriod(5000L);
+
+ UploadInfo part1Info = new UploadInfo();
+ part1Info.setLength(5L);
+ UploadInfo part1 = storageService.create(part1Info, null);
+ storageService.append(part1, new ByteArrayInputStream("part1".getBytes()));
+
+ UploadInfo finalInfo = new UploadInfo();
+ finalInfo.setConcatenationPartIds(Arrays.asList("/test/upload/" + part1.getId()));
+ finalInfo.setUploadType(UploadType.CONCATENATED);
+ UploadInfo createdFinal = storageService.create(finalInfo, null);
+
+ concatenationService.merge(createdFinal);
+
+ assertNotNull(createdFinal.getExpirationTimestamp());
+ assertTrue(createdFinal.getExpirationTimestamp() > System.currentTimeMillis());
+ }
+
+ @Test
+ public void mergeShouldDoNothingWhenIncompleteOrExpiredPartials() throws Exception {
+ UploadInfo part1Info = new UploadInfo();
+ part1Info.setLength(10L); // length 10, but offset 0 (in progress)
+ UploadInfo part1 = storageService.create(part1Info, null);
+
+ UploadInfo finalInfo = new UploadInfo();
+ finalInfo.setConcatenationPartIds(Arrays.asList("/test/upload/" + part1.getId()));
+ finalInfo.setUploadType(UploadType.CONCATENATED);
+ UploadInfo createdFinal = storageService.create(finalInfo, null);
+
+ concatenationService.merge(createdFinal);
+ // Should not merge since part1 is still in progress
+ assertEquals(Long.valueOf(0L), createdFinal.getOffset());
+ }
+
+ @Test
+ public void mergeShouldDoNothingWhenPartInfoLengthIsNull() throws Exception {
+ UploadInfo part1Info = new UploadInfo();
+ UploadInfo part1 = storageService.create(part1Info, null); // length null
+
+ UploadInfo finalInfo = new UploadInfo();
+ finalInfo.setConcatenationPartIds(Arrays.asList("/test/upload/" + part1.getId()));
+ finalInfo.setUploadType(UploadType.CONCATENATED);
+ UploadInfo createdFinal = storageService.create(finalInfo, null);
+
+ concatenationService.merge(createdFinal);
+ assertEquals(Long.valueOf(0L), createdFinal.getOffset());
+ }
+
+ @Test
+ public void mergeShouldDoNothingWhenFinalUploadNotInProgress() throws Exception {
+ UploadInfo finalInfo = new UploadInfo();
+ finalInfo.setOffset(10L);
+ finalInfo.setLength(10L); // upload completed, not in progress
+ finalInfo.setConcatenationPartIds(Arrays.asList("/test/upload/some-id"));
+
+ concatenationService.merge(finalInfo);
+ }
+
+ @Test
+ public void getConcatenatedBytesShouldReturnBytesForCompletedUpload() throws Exception {
+ UploadInfo info = new UploadInfo();
+ info.setOffset(5L);
+ info.setLength(5L); // completed (not in progress)
+ UploadInfo created = storageService.create(info, null);
+ storageService.append(created, new ByteArrayInputStream("hello".getBytes()));
+
+ InputStream is = concatenationService.getConcatenatedBytes(created);
+ assertNotNull(is);
+ assertEquals("hello", org.apache.commons.io.IOUtils.toString(is, StandardCharsets.UTF_8));
+ }
+
+ @Test
+ public void constructorPrefixSanitizationVariants() {
+ AzureBlobConcatenationService service1 =
+ new AzureBlobConcatenationService(containerClient, null, storageService);
+ AzureBlobConcatenationService service2 =
+ new AzureBlobConcatenationService(containerClient, "/custom/prefix", storageService);
+ AzureBlobConcatenationService service3 =
+ new AzureBlobConcatenationService(containerClient, "custom/prefix/", storageService);
+
+ assertNotNull(service1);
+ assertNotNull(service2);
+ assertNotNull(service3);
+ }
+
+ @Test
+ public void getConcatenatedBytesShouldReturnEmptyStreamForInProgressUploadWithIncompletePartials()
+ throws Exception {
+ UploadInfo part1Info = new UploadInfo();
+ part1Info.setLength(10L); // length 10, offset 0 (incomplete)
+ UploadInfo part1 = storageService.create(part1Info, null);
+
+ UploadInfo finalInfo = new UploadInfo();
+ finalInfo.setConcatenationPartIds(Arrays.asList("/test/upload/" + part1.getId()));
+ finalInfo.setUploadType(UploadType.CONCATENATED);
+ UploadInfo createdFinal = storageService.create(finalInfo, null);
+
+ InputStream is = concatenationService.getConcatenatedBytes(createdFinal);
+ assertNotNull(is);
+ assertEquals(0, is.available());
+ }
+
+ @Test
+ public void mergeWithEmptyPartIdsListShouldDoNothing() throws Exception {
+ UploadInfo finalInfo = new UploadInfo();
+ finalInfo.setConcatenationPartIds(java.util.Collections.emptyList());
+ finalInfo.setUploadType(UploadType.CONCATENATED);
+ UploadInfo createdFinal = storageService.create(finalInfo, null);
+
+ concatenationService.merge(createdFinal);
+ assertEquals(Long.valueOf(0L), createdFinal.getOffset());
+ }
+}
diff --git a/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobLockingService.java b/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobLockingService.java
new file mode 100644
index 0000000..f646e5a
--- /dev/null
+++ b/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobLockingService.java
@@ -0,0 +1,230 @@
+package me.desair.tus.server.upload.azure;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import com.azure.storage.blob.BlobContainerClient;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import me.desair.tus.server.TestUtils;
+import me.desair.tus.server.exception.UploadAlreadyLockedException;
+import me.desair.tus.server.upload.TimeBasedUploadIdFactory;
+import me.desair.tus.server.upload.UploadId;
+import me.desair.tus.server.upload.UploadLock;
+import me.desair.tus.server.util.InterruptibleInputStream;
+import org.junit.AfterClass;
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.testcontainers.containers.GenericContainer;
+
+public class ITAzureBlobLockingService {
+
+ private static GenericContainer azuriteContainer;
+
+ @BeforeClass
+ public static void setUpClass() {
+ Assume.assumeTrue(
+ "Container runtime is not available; skipping Testcontainers Azurite test",
+ TestUtils.isContainerRuntimeAvailable());
+ azuriteContainer = TestUtils.createAzuriteContainer();
+ azuriteContainer.start();
+ }
+
+ @AfterClass
+ public static void tearDownClass() {
+ if (azuriteContainer != null) {
+ azuriteContainer.stop();
+ }
+ }
+
+ private BlobContainerClient containerClient;
+ private AzureBlobLockingService lockingService;
+
+ @Before
+ public void setUp() {
+ Assume.assumeTrue(TestUtils.isContainerRuntimeAvailable());
+ containerClient =
+ TestUtils.createBlobContainerClient(
+ azuriteContainer, "lock-unit-container-" + System.nanoTime());
+ lockingService = new AzureBlobLockingService(containerClient);
+ TimeBasedUploadIdFactory idFactory = new TimeBasedUploadIdFactory();
+ idFactory.setUploadUri("/test/upload");
+ lockingService.setIdFactory(idFactory);
+ }
+
+ @Test
+ public void lockUploadByUriShouldReturnNullOnInvalidUri() throws Exception {
+ assertNull(lockingService.lockUploadByUri("invalid-uri-no-id"));
+ }
+
+ @Test
+ public void isLockedShouldReturnFalseWhenLockBlobDoesNotExist() {
+ assertFalse(lockingService.isLocked(new UploadId("12345")));
+ }
+
+ @Test
+ public void lockUploadByUriShouldAcquireLock() throws Exception {
+ UploadLock lock = lockingService.lockUploadByUri("/test/upload/12345");
+ assertNotNull(lock);
+ assertEquals("/test/upload/12345", lock.getUploadUri());
+ assertTrue(lockingService.isLocked(new UploadId("12345")));
+ lock.release();
+ }
+
+ @Test(expected = UploadAlreadyLockedException.class)
+ public void lockUploadByUriShouldThrowOnLockContention() throws Exception {
+ UploadLock lock1 = lockingService.lockUploadByUri("/test/upload/12345");
+ assertNotNull(lock1);
+ try {
+ lockingService.lockUploadByUri("/test/upload/12345");
+ } finally {
+ lock1.release();
+ }
+ }
+
+ @Test
+ public void registerInputStreamAndRequestReleaseShouldInterruptStream() throws Exception {
+ ByteArrayInputStream bais = new ByteArrayInputStream("data".getBytes());
+ InterruptibleInputStream stream = new InterruptibleInputStream(bais);
+
+ lockingService.registerInputStream("/test/upload/12345", stream);
+ lockingService.requestLockRelease("/test/upload/12345");
+
+ try {
+ stream.read();
+ } catch (Exception e) {
+ assertNotNull(e);
+ }
+ }
+
+ @Test
+ public void cleanupStaleLocksShouldNotThrow() throws Exception {
+ lockingService.cleanupStaleLocks();
+ }
+
+ @Test
+ public void closeShouldCleanUpResources() throws Exception {
+ lockingService.close();
+ lockingService.close();
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void constructorShouldThrowOnNullContainerClient() {
+ new AzureBlobLockingService(null);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void setIdFactoryShouldThrowOnNull() {
+ lockingService.setIdFactory(null);
+ }
+
+ @Test
+ public void isLockedShouldReturnFalseForNullId() {
+ assertFalse(lockingService.isLocked(null));
+ }
+
+ @Test
+ public void isLockedShouldReturnFalseWhenPropertiesThrowException() {
+ assertFalse(lockingService.isLocked(new UploadId("non-existent-lock-id-999")));
+ }
+
+ @Test
+ public void registerInputStreamShouldDoNothingOnInvalidUriOrStandardStream() {
+ ByteArrayInputStream bais = new ByteArrayInputStream("test".getBytes());
+
+ lockingService.registerInputStream("invalid-uri", bais);
+ lockingService.registerInputStream("/test/upload/12345", bais);
+ }
+
+ @Test
+ public void requestLockReleaseShouldDoNothingOnInvalidUri() {
+ lockingService.requestLockRelease("invalid-uri");
+ }
+
+ @Test
+ public void prefixSanitizationVariants() {
+ AzureBlobLockingService service1 = new AzureBlobLockingService(containerClient, null);
+ AzureBlobLockingService service2 =
+ new AzureBlobLockingService(containerClient, "/custom/locks");
+ AzureBlobLockingService service3 =
+ new AzureBlobLockingService(containerClient, "custom/locks/");
+
+ assertNotNull(service1);
+ assertNotNull(service2);
+ assertNotNull(service3);
+ }
+
+ @Test
+ public void watchdogPollingDetectsStopSignalBlob() throws Exception {
+ ByteArrayInputStream bais = new ByteArrayInputStream("data".getBytes());
+ InterruptibleInputStream stream = new InterruptibleInputStream(bais);
+
+ lockingService.registerInputStream("/test/upload/54321", stream);
+
+ com.azure.storage.blob.BlobClient stopBlob = containerClient.getBlobClient("locks/54321.stop");
+ stopBlob.upload(com.azure.core.util.BinaryData.fromString("stop"), true);
+
+ long deadline = System.currentTimeMillis() + 3500L;
+ while (!stream.isInterrupted() && System.currentTimeMillis() < deadline) {
+ Thread.sleep(100L);
+ }
+
+ assertTrue("Expected stream to be interrupted by watchdog thread", stream.isInterrupted());
+ assertFalse("Expected .stop blob to be deleted by watchdog thread", stopBlob.exists());
+ }
+
+ @Test
+ public void ensureLockBlobExistsHandlesExceptions() {
+ com.azure.storage.blob.BlobClient lockBlob =
+ containerClient.getBlobClient("locks/nonexistentcontainer/invalid.lock");
+ lockingService.ensureLockBlobExists(lockBlob);
+ }
+
+ @Test
+ public void closeInterruptsActiveWatchdogThread() throws Exception {
+ ByteArrayInputStream bais = new ByteArrayInputStream("data".getBytes());
+ InterruptibleInputStream stream = new InterruptibleInputStream(bais);
+
+ lockingService.registerInputStream("/test/upload/88888", stream);
+ lockingService.close();
+ }
+
+ @Test(expected = IOException.class)
+ public void lockUploadByUriShouldThrowIOExceptionOnStorageException() throws Exception {
+ com.azure.storage.blob.BlobServiceClient serviceClient = containerClient.getServiceClient();
+ BlobContainerClient nonExistentContainer =
+ serviceClient.getBlobContainerClient(
+ "non-existent-container-" + System.currentTimeMillis());
+ AzureBlobLockingService service = new AzureBlobLockingService(nonExistentContainer);
+ TimeBasedUploadIdFactory idFactory = new TimeBasedUploadIdFactory();
+ idFactory.setUploadUri("/test/upload");
+ service.setIdFactory(idFactory);
+
+ service.lockUploadByUri("/test/upload/12345");
+ }
+
+ @Test
+ public void testCreateAndDeleteStopSignalBlobExceptionHandling() {
+ com.azure.storage.blob.BlobServiceClient serviceClient = containerClient.getServiceClient();
+ BlobContainerClient invalidContainer =
+ serviceClient.getBlobContainerClient("invalid-container-" + System.currentTimeMillis());
+
+ AzureBlobLockingService invalidLocking = new AzureBlobLockingService(invalidContainer);
+ TimeBasedUploadIdFactory idFactory = new TimeBasedUploadIdFactory();
+ idFactory.setUploadUri("/test/upload");
+ invalidLocking.setIdFactory(idFactory);
+
+ ByteArrayInputStream bais = new ByteArrayInputStream("data".getBytes());
+ InterruptibleInputStream stream = new InterruptibleInputStream(bais);
+ invalidLocking.registerInputStream("/test/upload/12345", stream);
+
+ // Requesting lock release triggers createStopSignalBlob & deleteStopSignalBlob on invalid
+ // container
+ invalidLocking.requestLockRelease("/test/upload/12345");
+ }
+}
diff --git a/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobRufhProtocol.java b/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobRufhProtocol.java
new file mode 100644
index 0000000..02a9d47
--- /dev/null
+++ b/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobRufhProtocol.java
@@ -0,0 +1,59 @@
+package me.desair.tus.server.upload.azure;
+
+import com.azure.storage.blob.BlobContainerClient;
+import me.desair.tus.server.AbstractITRufhProtocol;
+import me.desair.tus.server.TestUtils;
+import me.desair.tus.server.TusFileUploadService;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.testcontainers.containers.GenericContainer;
+
+/**
+ * End-to-end integration test suite verifying the IETF Resumable Uploads for HTTP (RUFH) protocol
+ * implementation backed by {@link AzureBlobStorageService} and {@link AzureBlobLockingService} on
+ * Azurite using the official Azure Storage Blob SDK.
+ */
+public class ITAzureBlobRufhProtocol extends AbstractITRufhProtocol {
+
+ private static GenericContainer azurite;
+ private static BlobContainerClient containerClient;
+ private static final String CONTAINER = "test-rufh-azure-container";
+
+ @BeforeClass
+ public static void setUpClass() {
+ org.junit.Assume.assumeTrue(
+ "Container runtime is not available; skipping Testcontainers Azurite test",
+ TestUtils.isContainerRuntimeAvailable());
+
+ azurite = TestUtils.createAzuriteContainer();
+ azurite.start();
+
+ containerClient = TestUtils.createBlobContainerClient(azurite, CONTAINER);
+ }
+
+ @AfterClass
+ public static void tearDownClass() {
+ if (azurite != null) {
+ azurite.stop();
+ }
+ }
+
+ @Override
+ protected TusFileUploadService createTusFileUploadService() {
+ org.junit.Assume.assumeTrue(TestUtils.isContainerRuntimeAvailable());
+
+ AzureBlobStorageService azureStorage = new AzureBlobStorageService(containerClient);
+ AzureBlobLockingService azureLocking = new AzureBlobLockingService(containerClient);
+ AzureBlobConcatenationService azureConcat =
+ new AzureBlobConcatenationService(containerClient, azureStorage);
+ azureStorage.setUploadConcatenationService(azureConcat);
+
+ return new TusFileUploadService()
+ .withUploadUri(UPLOAD_URI)
+ .withUploadStorageService(azureStorage)
+ .withUploadLockingService(azureLocking)
+ .withMaxUploadSize(1073741824L)
+ .withUploadExpirationPeriod(2L * 24 * 60 * 60 * 1000)
+ .withDownloadFeature();
+ }
+}
diff --git a/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobStorageService.java b/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobStorageService.java
new file mode 100644
index 0000000..b3ea607
--- /dev/null
+++ b/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobStorageService.java
@@ -0,0 +1,530 @@
+package me.desair.tus.server.upload.azure;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import com.azure.storage.blob.BlobContainerClient;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import me.desair.tus.server.TestUtils;
+import me.desair.tus.server.checksum.ChecksumAlgorithm;
+import me.desair.tus.server.exception.MaxAppendSizeExceededException;
+import me.desair.tus.server.exception.MinAppendSizeNotMetException;
+import me.desair.tus.server.exception.UploadNotFoundException;
+import me.desair.tus.server.upload.UploadId;
+import me.desair.tus.server.upload.UploadInfo;
+import me.desair.tus.server.upload.UploadLock;
+import org.junit.AfterClass;
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.testcontainers.containers.GenericContainer;
+
+public class ITAzureBlobStorageService {
+
+ private static GenericContainer azuriteContainer;
+
+ @BeforeClass
+ public static void setUpClass() {
+ Assume.assumeTrue(
+ "Container runtime is not available; skipping Testcontainers Azurite test",
+ TestUtils.isContainerRuntimeAvailable());
+ azuriteContainer = TestUtils.createAzuriteContainer();
+ azuriteContainer.start();
+ }
+
+ @AfterClass
+ public static void tearDownClass() {
+ if (azuriteContainer != null) {
+ azuriteContainer.stop();
+ }
+ }
+
+ private BlobContainerClient containerClient;
+ private AzureBlobStorageService storageService;
+
+ @Before
+ public void setUp() {
+ Assume.assumeTrue(TestUtils.isContainerRuntimeAvailable() && azuriteContainer != null);
+ containerClient =
+ TestUtils.createBlobContainerClient(
+ azuriteContainer, "unit-test-container-" + System.nanoTime());
+ storageService = new AzureBlobStorageService(containerClient);
+ }
+
+ @Test
+ public void configurationGettersAndSetters() {
+ storageService.setMaxAppendSize(500L);
+ assertEquals(Long.valueOf(500L), storageService.getMaxAppendSize());
+
+ storageService.setMinAppendSize(100L);
+ assertEquals(Long.valueOf(100L), storageService.getMinAppendSize());
+
+ storageService.setPreferredBlockSize(8L * 1024 * 1024);
+ assertEquals(8L * 1024 * 1024, storageService.getPreferredBlockSize());
+
+ storageService.setUploadDeduplicationEnabled(true);
+ assertTrue(storageService.isUploadDeduplicationEnabled());
+
+ storageService.setUploadExpirationPeriod(3600000L);
+ assertEquals(Long.valueOf(3600000L), storageService.getUploadExpirationPeriod());
+ }
+
+ @Test
+ public void constructorCustomPrefixes() {
+ AzureBlobStorageService customService =
+ new AzureBlobStorageService(
+ containerClient,
+ "custom-uploads/",
+ "custom-metadata/",
+ "custom-checksums/",
+ "custom-parts/",
+ java.nio.file.Paths.get(System.getProperty("java.io.tmpdir")));
+ assertNotNull(customService);
+ }
+
+ @Test
+ public void getAzureBlobNameShouldReturnBlobName() throws Exception {
+ UploadInfo info = new UploadInfo();
+ info.setId(new UploadId("12345"));
+ assertEquals("uploads/12345", storageService.getAzureBlobName(info));
+
+ info.setDuplicatesUploadId(new UploadId("parent-999"));
+ assertEquals("uploads/parent-999", storageService.getAzureBlobName(info));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void setPreferredBlockSizeTooSmallShouldThrow() {
+ storageService.setPreferredBlockSize(1024L); // less than 4MB
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void setPreferredBlockSizeTooLargeShouldThrow() {
+ storageService.setPreferredBlockSize(5000L * 1024 * 1024); // greater than 4000MB
+ }
+
+ @Test(expected = MinAppendSizeNotMetException.class)
+ public void minAppendSizeNotMetShouldThrow() throws Exception {
+ storageService.setMinAppendSize(100L);
+
+ UploadInfo info = new UploadInfo();
+ info.setLength(500L);
+ UploadInfo created = storageService.create(info, "owner1");
+
+ storageService.append(created, new ByteArrayInputStream("0123456789".getBytes()));
+ }
+
+ @Test(expected = MaxAppendSizeExceededException.class)
+ public void appendExceedsMaxAppendSizeShouldThrow() throws Exception {
+ storageService.setMaxAppendSize(5L);
+
+ UploadInfo info = new UploadInfo();
+ info.setLength(500L);
+ UploadInfo created = storageService.create(info, "owner1");
+
+ storageService.append(created, new ByteArrayInputStream("0123456789".getBytes()));
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void removeLastNumberOfBytesShouldThrowOnNullInfo() throws Exception {
+ storageService.removeLastNumberOfBytes(null, 10L);
+ }
+
+ @Test(expected = UploadNotFoundException.class)
+ public void getUploadedBytesByUriNotFoundShouldThrow() throws Exception {
+ storageService.getUploadedBytes("/test/upload/non-existent", "owner1");
+ }
+
+ @Test
+ public void getUploadedBytesNullIdShouldReturnNull() throws Exception {
+ assertNull(storageService.getUploadedBytes((UploadId) null));
+ }
+
+ @Test(expected = UploadNotFoundException.class)
+ public void copyUploadToNotFoundShouldThrow() throws Exception {
+ UploadInfo info = new UploadInfo();
+ info.setId(new UploadId("non-existent-id"));
+ storageService.copyUploadTo(info, new ByteArrayOutputStream());
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void copyUploadToNullInfoShouldThrow() throws Exception {
+ storageService.copyUploadTo(null, new ByteArrayOutputStream());
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void copyUploadToNullStreamShouldThrow() throws Exception {
+ storageService.copyUploadTo(new UploadInfo(), null);
+ }
+
+ @Test
+ public void terminateUploadNullInfoShouldDoNothing() throws Exception {
+ storageService.terminateUpload(null);
+ storageService.terminateUpload(new UploadInfo());
+ }
+
+ @Test
+ public void getAzureBlobNameNullInfoShouldReturnNull() throws Exception {
+ assertNull(storageService.getAzureBlobName((UploadInfo) null));
+ assertNull(storageService.getAzureBlobName("/test/upload/invalid", "owner1"));
+ }
+
+ @Test
+ public void getUploadInfoByChecksumDisabledOrNull() throws Exception {
+ storageService.setUploadDeduplicationEnabled(false);
+ assertNull(storageService.getUploadInfoByChecksum("checksum", ChecksumAlgorithm.MD5));
+
+ storageService.setUploadDeduplicationEnabled(true);
+ assertNull(storageService.getUploadInfoByChecksum(null, ChecksumAlgorithm.MD5));
+ assertNull(storageService.getUploadInfoByChecksum("checksum", null));
+ }
+
+ @Test
+ public void createZeroLengthUploadShouldCommitEmptyDataBlob() throws Exception {
+ UploadInfo info = new UploadInfo();
+ info.setLength(0L);
+ UploadInfo created = storageService.create(info, "owner1");
+
+ assertNotNull(created.getId());
+ assertEquals(Long.valueOf(0L), created.getOffset());
+
+ UploadInfo fetched = storageService.getUploadInfo(created.getId());
+ assertNotNull(fetched);
+ assertEquals(Long.valueOf(0L), fetched.getOffset());
+ }
+
+ @Test
+ public void appendConsecutiveSubThresholdChunksToPartBlob() throws Exception {
+ storageService.setPreferredBlockSize(4L * 1024 * 1024); // 4MB
+
+ UploadInfo info = new UploadInfo();
+ info.setLength(100L);
+ UploadInfo created = storageService.create(info, "owner1");
+
+ storageService.append(created, new ByteArrayInputStream("0123456789".getBytes()));
+ assertEquals(Long.valueOf(10L), created.getOffset());
+
+ storageService.append(created, new ByteArrayInputStream("abcdefghij".getBytes()));
+ assertEquals(Long.valueOf(20L), created.getOffset());
+
+ try (InputStream is = storageService.getUploadedBytes(created.getId())) {
+ assertEquals(
+ "0123456789abcdefghij",
+ org.apache.commons.io.IOUtils.toString(is, StandardCharsets.UTF_8));
+ }
+ }
+
+ @Test
+ public void maxAppendSizeFallback() throws Exception {
+ UploadInfo info = new UploadInfo();
+ info.setLength(100L);
+ UploadInfo created = storageService.create(info, "owner1");
+ storageService.setMaxAppendSize(null); // disable limit
+
+ storageService.append(created, new ByteArrayInputStream("0123456789".getBytes()));
+ assertEquals(Long.valueOf(10L), created.getOffset());
+ }
+
+ @Test
+ public void getUploadedBytesShouldReturnInputStream() throws Exception {
+ UploadInfo info = new UploadInfo();
+ info.setLength(12L);
+ UploadInfo created = storageService.create(info, "owner1");
+
+ storageService.append(created, new ByteArrayInputStream("Hello World!".getBytes()));
+
+ try (InputStream is = storageService.getUploadedBytes(created.getId())) {
+ assertNotNull(is);
+ assertEquals(
+ "Hello World!", org.apache.commons.io.IOUtils.toString(is, StandardCharsets.UTF_8));
+ }
+
+ try (InputStream is =
+ storageService.getUploadedBytes("/test/upload/" + created.getId(), "owner1")) {
+ assertNotNull(is);
+ assertEquals(
+ "Hello World!", org.apache.commons.io.IOUtils.toString(is, StandardCharsets.UTF_8));
+ }
+ }
+
+ @Test
+ public void getUploadInfoShouldReturnInfo() throws Exception {
+ UploadInfo info = new UploadInfo();
+ info.setLength(100L);
+ UploadInfo created = storageService.create(info, "owner1");
+
+ UploadInfo fetchedById = storageService.getUploadInfo(created.getId());
+ assertNotNull(fetchedById);
+ assertEquals(created.getId(), fetchedById.getId());
+
+ UploadInfo fetchedByUri =
+ storageService.getUploadInfo("/test/upload/" + created.getId(), "owner1");
+ assertNotNull(fetchedByUri);
+ assertEquals(created.getId(), fetchedByUri.getId());
+
+ assertNull(storageService.getUploadInfo(new UploadId("non-existing-id")));
+ assertNull(storageService.getUploadInfo("/test/upload/non-existing-id", "owner1"));
+ }
+
+ @Test
+ public void copyUploadToShouldCopyDataToOutputStream() throws Exception {
+ UploadInfo info = new UploadInfo();
+ info.setLength(11L);
+ UploadInfo created = storageService.create(info, "owner1");
+
+ storageService.append(created, new ByteArrayInputStream("Hello Azure".getBytes()));
+
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ storageService.copyUploadTo(created, baos);
+ assertEquals("Hello Azure", baos.toString());
+
+ ByteArrayOutputStream baosUri = new ByteArrayOutputStream();
+ UploadInfo fetchedUri =
+ storageService.getUploadInfo("/test/upload/" + created.getId(), "owner1");
+ storageService.copyUploadTo(fetchedUri, baosUri);
+ assertEquals("Hello Azure", baosUri.toString());
+ }
+
+ @Test
+ public void terminateUploadShouldDeleteBlobs() throws Exception {
+ UploadInfo info = new UploadInfo();
+ info.setLength(10L);
+ UploadInfo created = storageService.create(info, "owner1");
+
+ storageService.append(created, new ByteArrayInputStream("0123456789".getBytes()));
+ assertNotNull(storageService.getUploadInfo(created.getId()));
+
+ storageService.terminateUpload(created);
+ assertNull(storageService.getUploadInfo(created.getId()));
+ }
+
+ @Test
+ public void removeLastNumberOfBytesPartBlobOnly() throws Exception {
+ storageService.setPreferredBlockSize(4L * 1024 * 1024);
+
+ UploadInfo info = new UploadInfo();
+ info.setLength(100L);
+ UploadInfo created = storageService.create(info, "owner1");
+
+ storageService.append(created, new ByteArrayInputStream("0123456789".getBytes()));
+ assertEquals(Long.valueOf(10L), created.getOffset());
+
+ // Remove 4 bytes from sub-threshold part blob
+ storageService.removeLastNumberOfBytes(created, 4L);
+ assertEquals(Long.valueOf(6L), created.getOffset());
+
+ try (InputStream is = storageService.getUploadedBytes(created.getId())) {
+ assertNotNull(is);
+ assertEquals("012345", org.apache.commons.io.IOUtils.toString(is, StandardCharsets.UTF_8));
+ }
+
+ // Remove remaining bytes
+ storageService.removeLastNumberOfBytes(created, 6L);
+ assertEquals(Long.valueOf(0L), created.getOffset());
+ }
+
+ @Test
+ public void removeLastNumberOfBytesTrimPartBlobWithCommittedBlocks() throws Exception {
+ storageService.setPreferredBlockSize(4L * 1024 * 1024); // 4MB blocks
+
+ byte[] data = new byte[9 * 1024 * 1024];
+ java.util.Arrays.fill(data, (byte) 'B');
+
+ UploadInfo info = new UploadInfo();
+ info.setLength((long) data.length);
+ UploadInfo created = storageService.create(info, "owner1");
+
+ storageService.append(created, new ByteArrayInputStream(data));
+ assertEquals(Long.valueOf(data.length), created.getOffset());
+
+ // Truncate by 500KB (targetOffset 8.5MB > 8MB block blob size)
+ storageService.removeLastNumberOfBytes(created, 500 * 1024);
+ assertEquals(Long.valueOf(data.length - 500 * 1024), created.getOffset());
+
+ try (InputStream is = storageService.getUploadedBytes(created.getId())) {
+ assertNotNull(is);
+ byte[] readBytes = org.apache.commons.io.IOUtils.toByteArray(is);
+ assertEquals(data.length - 500 * 1024, readBytes.length);
+ }
+ }
+
+ @Test
+ public void appendMultiBlockPayloadAndTruncateCommittedBlocks() throws Exception {
+ storageService.setPreferredBlockSize(4L * 1024 * 1024); // 4MB blocks
+
+ byte[] data = new byte[9 * 1024 * 1024];
+ java.util.Arrays.fill(data, (byte) 'A');
+
+ UploadInfo info = new UploadInfo();
+ info.setLength((long) data.length);
+ UploadInfo created = storageService.create(info, "owner1");
+
+ storageService.append(created, new ByteArrayInputStream(data));
+ assertEquals(Long.valueOf(data.length), created.getOffset());
+
+ // Truncate by 2MB (targetOffset 7MB <= 8MB block blob size)
+ storageService.removeLastNumberOfBytes(created, 2L * 1024 * 1024);
+ assertEquals(Long.valueOf(7L * 1024 * 1024), created.getOffset());
+
+ try (InputStream is = storageService.getUploadedBytes(created.getId())) {
+ assertNotNull(is);
+ }
+ }
+
+ @Test
+ public void cleanupExpiredUploadsShouldDeleteExpired() throws Exception {
+ storageService.setUploadExpirationPeriod(1L);
+
+ UploadInfo info = new UploadInfo();
+ info.setLength(10L);
+ UploadInfo created = storageService.create(info, "owner1");
+
+ storageService.append(created, new ByteArrayInputStream("0123456789".getBytes()));
+
+ Thread.sleep(50L);
+ // Cleanup with expiration period 0 (all uploads expired)
+ storageService.cleanupExpiredUploads(new AzureBlobLockingService(containerClient));
+ assertNull(storageService.getUploadInfo(created.getId()));
+ }
+
+ @Test
+ public void getUploadInfoByChecksumSelfCleaningStaleIndex() throws Exception {
+ storageService.setUploadDeduplicationEnabled(true);
+
+ UploadInfo info = new UploadInfo();
+ info.setLength(5L);
+ info.setChecksum("5d41402abc4b2a76b9719d911017c592");
+ info.setChecksumAlgorithm(ChecksumAlgorithm.MD5);
+
+ UploadInfo created = storageService.create(info, "owner1");
+ storageService.append(created, new ByteArrayInputStream("hello".getBytes()));
+
+ UploadInfo found =
+ storageService.getUploadInfoByChecksum(
+ "5d41402abc4b2a76b9719d911017c592", ChecksumAlgorithm.MD5);
+ assertNotNull(found);
+ assertEquals(created.getId(), found.getId());
+
+ // Manually delete upload blobs to simulate stale index
+ storageService.terminateUpload(created);
+
+ // Stale index lookup should self-clean and return null
+ assertNull(
+ storageService.getUploadInfoByChecksum(
+ "5d41402abc4b2a76b9719d911017c592", ChecksumAlgorithm.MD5));
+ }
+
+ @Test
+ public void testFullUploadLifecycleOnAzurite() throws Exception {
+ AzureBlobStorageService storage = new AzureBlobStorageService(containerClient);
+ AzureBlobLockingService locking = new AzureBlobLockingService(containerClient);
+
+ UploadInfo info = new UploadInfo();
+ info.setLength(11L);
+
+ UploadInfo created = storage.create(info, "owner1");
+ assertNotNull(created.getId());
+
+ try (UploadLock lock = locking.lockUploadByUri("/test/upload/" + created.getId())) {
+ assertNotNull(lock);
+ storage.append(created, new ByteArrayInputStream("hello ".getBytes()));
+ storage.append(created, new ByteArrayInputStream("world".getBytes()));
+ }
+
+ UploadInfo fetched = storage.getUploadInfo(created.getId());
+ assertNotNull(fetched);
+ assertEquals(Long.valueOf(11L), fetched.getOffset());
+
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ storage.copyUploadTo(fetched, baos);
+ assertEquals("hello world", baos.toString());
+
+ String blobName = storage.getAzureBlobName("/test/upload/" + created.getId(), "owner1");
+ assertEquals("uploads/" + created.getId(), blobName);
+ }
+
+ @Test
+ public void testTruncateBytesOnAzurite() throws Exception {
+ AzureBlobStorageService storage = new AzureBlobStorageService(containerClient);
+
+ UploadInfo info = new UploadInfo();
+ info.setLength(10L);
+ UploadInfo created = storage.create(info, "owner1");
+
+ storage.append(created, new ByteArrayInputStream("0123456789".getBytes()));
+ assertEquals(Long.valueOf(10L), created.getOffset());
+
+ storage.removeLastNumberOfBytes(created, 3L);
+ assertEquals(Long.valueOf(7L), created.getOffset());
+
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ storage.copyUploadTo(created, baos);
+ assertEquals("0123456", baos.toString());
+ }
+
+ @Test
+ public void testDeduplicationOnAzurite() throws Exception {
+ AzureBlobStorageService storage = new AzureBlobStorageService(containerClient);
+ storage.setUploadDeduplicationEnabled(true);
+
+ UploadInfo parent = new UploadInfo();
+ parent.setLength(5L);
+ parent.setChecksum("5d41402abc4b2a76b9719d911017c592");
+ parent.setChecksumAlgorithm(ChecksumAlgorithm.MD5);
+ parent = storage.create(parent, "owner1");
+ storage.append(parent, new ByteArrayInputStream("hello".getBytes()));
+
+ UploadInfo child = new UploadInfo();
+ child.setLength(5L);
+ child.setChecksum("5d41402abc4b2a76b9719d911017c592");
+ child.setChecksumAlgorithm(ChecksumAlgorithm.MD5);
+ child = storage.create(child, "owner1");
+ storage.append(child, new ByteArrayInputStream("hello".getBytes()));
+
+ assertNotNull(child.getDuplicatesUploadId());
+ assertEquals(parent.getId(), child.getDuplicatesUploadId());
+
+ storage.terminateUpload(parent);
+ assertNull(
+ storage.getUploadInfoByChecksum("5d41402abc4b2a76b9719d911017c592", ChecksumAlgorithm.MD5));
+ }
+
+ @Test
+ public void appendConsecutiveCommittedBlocksShouldRetrieveCommittedBlockIds() throws Exception {
+ storageService.setPreferredBlockSize(4L * 1024 * 1024); // 4MB block size
+
+ byte[] block1 = new byte[4 * 1024 * 1024];
+ java.util.Arrays.fill(block1, (byte) 'X');
+
+ byte[] block2 = new byte[4 * 1024 * 1024];
+ java.util.Arrays.fill(block2, (byte) 'Y');
+
+ UploadInfo info = new UploadInfo();
+ info.setLength(10L * 1024 * 1024);
+ UploadInfo created = storageService.create(info, "owner1");
+
+ // 1st append commits block 1 (4MB)
+ storageService.append(created, new ByteArrayInputStream(block1));
+ assertEquals(Long.valueOf(4L * 1024 * 1024), created.getOffset());
+
+ // 2nd append calls getCommittedBlockIds(blockBlobClient) and commits block 2 (4MB)
+ storageService.append(created, new ByteArrayInputStream(block2));
+ assertEquals(Long.valueOf(8L * 1024 * 1024), created.getOffset());
+
+ // 3rd append calls getCommittedBlockIds(blockBlobClient) which retrieves 2 committed block IDs
+ // from Azurite
+ storageService.append(created, new ByteArrayInputStream("extra".getBytes()));
+ assertEquals(Long.valueOf(8L * 1024 * 1024 + 5), created.getOffset());
+
+ try (InputStream is = storageService.getUploadedBytes(created.getId())) {
+ assertNotNull(is);
+ byte[] readBytes = org.apache.commons.io.IOUtils.toByteArray(is);
+ assertEquals(8 * 1024 * 1024 + 5, readBytes.length);
+ }
+ }
+}
diff --git a/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobTusFileUploadService.java b/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobTusFileUploadService.java
new file mode 100644
index 0000000..6a9f9d5
--- /dev/null
+++ b/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobTusFileUploadService.java
@@ -0,0 +1,68 @@
+package me.desair.tus.server.upload.azure;
+
+import com.azure.storage.blob.BlobContainerClient;
+import me.desair.tus.server.AbstractITTusFileUploadService;
+import me.desair.tus.server.ProtocolVersion;
+import me.desair.tus.server.TestUtils;
+import me.desair.tus.server.TusFileUploadService;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.testcontainers.containers.GenericContainer;
+
+/**
+ * End-to-end integration test suite verifying {@link TusFileUploadService} backed by {@link
+ * AzureBlobStorageService} and {@link AzureBlobLockingService} on Azurite using the official Azure
+ * Storage Blob SDK. Extends {@link AbstractITTusFileUploadService} to run all Tus 1.0.0 protocol
+ * use cases against Azure Blob Storage.
+ */
+public class ITAzureBlobTusFileUploadService extends AbstractITTusFileUploadService {
+
+ private static GenericContainer azurite;
+ private static BlobContainerClient containerClient;
+ private static final String CONTAINER = "test-tus-azure-container";
+
+ @BeforeClass
+ public static void setUpClass() {
+ org.junit.Assume.assumeTrue(
+ "Container runtime is not available; skipping Testcontainers Azurite test",
+ TestUtils.isContainerRuntimeAvailable());
+
+ azurite = TestUtils.createAzuriteContainer();
+ azurite.start();
+
+ containerClient = TestUtils.createBlobContainerClient(azurite, CONTAINER);
+ }
+
+ @AfterClass
+ public static void tearDownClass() {
+ if (azurite != null) {
+ azurite.stop();
+ }
+ }
+
+ @Override
+ protected TusFileUploadService createTusFileUploadService() {
+ return createTusFileUploadService(UPLOAD_URI);
+ }
+
+ @Override
+ protected TusFileUploadService createTusFileUploadService(String uploadUri) {
+ org.junit.Assume.assumeTrue(TestUtils.isContainerRuntimeAvailable());
+
+ AzureBlobStorageService azureStorage = new AzureBlobStorageService(containerClient);
+ AzureBlobLockingService azureLocking = new AzureBlobLockingService(containerClient);
+ AzureBlobConcatenationService azureConcat =
+ new AzureBlobConcatenationService(containerClient, azureStorage);
+ azureStorage.setUploadConcatenationService(azureConcat);
+
+ return new TusFileUploadService()
+ .withUploadUri(uploadUri)
+ .withUploadStorageService(azureStorage)
+ .withUploadLockingService(azureLocking)
+ .withMaxUploadSize(1073741824L)
+ .withUploadExpirationPeriod(2L * 24 * 60 * 60 * 1000)
+ .withSupportedProtocolVersions(ProtocolVersion.TUS_1_0_0)
+ .withDownloadFeature()
+ .withChunkedTransferDecoding(true);
+ }
+}
diff --git a/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobUploadLock.java b/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobUploadLock.java
new file mode 100644
index 0000000..021f8c3
--- /dev/null
+++ b/src/test/java/me/desair/tus/server/upload/azure/ITAzureBlobUploadLock.java
@@ -0,0 +1,132 @@
+package me.desair.tus.server.upload.azure;
+
+import static org.junit.Assert.assertEquals;
+
+import com.azure.storage.blob.BlobClient;
+import com.azure.storage.blob.BlobContainerClient;
+import com.azure.storage.blob.specialized.BlobLeaseClient;
+import com.azure.storage.blob.specialized.BlobLeaseClientBuilder;
+import me.desair.tus.server.TestUtils;
+import me.desair.tus.server.upload.UploadLock;
+import org.junit.AfterClass;
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.testcontainers.containers.GenericContainer;
+
+public class ITAzureBlobUploadLock {
+
+ private static GenericContainer azuriteContainer;
+
+ @BeforeClass
+ public static void setUpClass() {
+ Assume.assumeTrue(
+ "Container runtime is not available; skipping Testcontainers Azurite test",
+ TestUtils.isContainerRuntimeAvailable());
+ azuriteContainer = TestUtils.createAzuriteContainer();
+ azuriteContainer.start();
+ }
+
+ @AfterClass
+ public static void tearDownClass() {
+ if (azuriteContainer != null) {
+ azuriteContainer.stop();
+ }
+ }
+
+ private BlobContainerClient containerClient;
+ private String uploadUri;
+ private UploadLock uploadLock;
+
+ @Before
+ public void setUp() throws Exception {
+ Assume.assumeTrue(TestUtils.isContainerRuntimeAvailable());
+ containerClient =
+ TestUtils.createBlobContainerClient(
+ azuriteContainer, "lock-unit-container2-" + System.nanoTime());
+
+ uploadUri = "/test/upload/12345";
+ BlobClient lockBlob = containerClient.getBlobClient("locks/12345.lock");
+ lockBlob.getAppendBlobClient().create();
+ BlobLeaseClient leaseClient = new BlobLeaseClientBuilder().blobClient(lockBlob).buildClient();
+ leaseClient.acquireLease(30);
+
+ uploadLock = new AzureBlobUploadLock(leaseClient, lockBlob, uploadUri);
+ }
+
+ @Test
+ public void getUploadUriShouldReturnUri() {
+ assertEquals(uploadUri, uploadLock.getUploadUri());
+ }
+
+ @Test
+ public void releaseShouldReleaseLease() {
+ uploadLock.release();
+ }
+
+ @Test
+ public void closeShouldCallRelease() throws Exception {
+ uploadLock.close();
+ }
+
+ @Test
+ public void doubleReleaseShouldBeIdempotent() {
+ uploadLock.release();
+ uploadLock.release();
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void constructorShouldThrowOnNullLeaseClient() {
+ BlobClient lockBlob = containerClient.getBlobClient("locks/nulltest.lock");
+ new AzureBlobUploadLock(null, lockBlob, "/uri");
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void constructorShouldThrowOnNullLockBlob() {
+ BlobClient lockBlob = containerClient.getBlobClient("locks/nulltest2.lock");
+ lockBlob.getAppendBlobClient().create();
+ BlobLeaseClient leaseClient = new BlobLeaseClientBuilder().blobClient(lockBlob).buildClient();
+ new AzureBlobUploadLock(leaseClient, null, "/uri");
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void constructorShouldThrowOnNullUploadUri() {
+ BlobClient lockBlob = containerClient.getBlobClient("locks/nulltest3.lock");
+ lockBlob.getAppendBlobClient().create();
+ BlobLeaseClient leaseClient = new BlobLeaseClientBuilder().blobClient(lockBlob).buildClient();
+ new AzureBlobUploadLock(leaseClient, lockBlob, null);
+ }
+
+ @Test
+ public void releaseShouldHandleReleaseLeaseExceptionGracefully() {
+ BlobClient lockBlob = containerClient.getBlobClient("locks/releasetest.lock");
+ lockBlob.getAppendBlobClient().create();
+ BlobLeaseClient leaseClient = new BlobLeaseClientBuilder().blobClient(lockBlob).buildClient();
+ AzureBlobUploadLock lock = new AzureBlobUploadLock(leaseClient, lockBlob, "/test/uri");
+ lock.release();
+ }
+
+ @Test
+ public void renewLeaseWhenReleasedShouldReturnImmediately() {
+ AzureBlobUploadLock lock = (AzureBlobUploadLock) uploadLock;
+ lock.release();
+ lock.renewLease();
+ }
+
+ @Test
+ public void renewLeaseShouldRenewActiveLease() {
+ AzureBlobUploadLock lock = (AzureBlobUploadLock) uploadLock;
+ lock.renewLease();
+ }
+
+ @Test
+ public void renewLeaseFailureShouldCatchExceptionAndSetReleased() {
+ BlobClient lockBlob = containerClient.getBlobClient("locks/renewfail.lock");
+ lockBlob.getAppendBlobClient().create();
+ BlobLeaseClient leaseClient = new BlobLeaseClientBuilder().blobClient(lockBlob).buildClient();
+ AzureBlobUploadLock lock = new AzureBlobUploadLock(leaseClient, lockBlob, "/test/uri");
+ lock.renewLease();
+ lock.renewLease();
+ }
+}
diff --git a/src/test/java/me/desair/tus/server/upload/cache/ThreadLocalCachedStorageAndLockingServiceTest.java b/src/test/java/me/desair/tus/server/upload/cache/ThreadLocalCachedStorageAndLockingServiceTest.java
index 49ce17b..d308cb8 100644
--- a/src/test/java/me/desair/tus/server/upload/cache/ThreadLocalCachedStorageAndLockingServiceTest.java
+++ b/src/test/java/me/desair/tus/server/upload/cache/ThreadLocalCachedStorageAndLockingServiceTest.java
@@ -215,5 +215,8 @@ public void testDelegateMethods() throws Exception {
service.requestLockRelease("/files/1");
verify(mockLocking, times(1)).requestLockRelease("/files/1");
+
+ service.close();
+ verify(mockLocking, times(1)).close();
}
}
diff --git a/src/test/java/me/desair/tus/server/upload/disk/DiskLockingServiceTest.java b/src/test/java/me/desair/tus/server/upload/disk/DiskLockingServiceTest.java
index 3dc9bac..98e7d07 100644
--- a/src/test/java/me/desair/tus/server/upload/disk/DiskLockingServiceTest.java
+++ b/src/test/java/me/desair/tus/server/upload/disk/DiskLockingServiceTest.java
@@ -23,6 +23,7 @@
import me.desair.tus.server.util.InterruptibleInputStream;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
+import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
@@ -79,10 +80,17 @@ public UploadId answer(InvocationOnMock invocation) throws Throwable {
lockingService = new DiskLockingService(idFactory, storagePath.toString());
}
+ @After
+ public void tearDown() throws Exception {
+ if (lockingService != null) {
+ lockingService.close();
+ }
+ }
+
@Test
public void lockUploadByUri() throws Exception {
- UploadLock uploadLock =
- lockingService.lockUploadByUri("/upload/test/000003f1-a850-49de-af03-997272d834c9");
+ String uploadIdStr = UUID.randomUUID().toString();
+ UploadLock uploadLock = lockingService.lockUploadByUri("/upload/test/" + uploadIdStr);
assertThat(uploadLock, not(nullValue()));
@@ -91,23 +99,21 @@ public void lockUploadByUri() throws Exception {
@Test
public void isLockedTrue() throws Exception {
- UploadLock uploadLock =
- lockingService.lockUploadByUri("/upload/test/000003f1-a850-49de-af03-997272d834c9");
+ String uploadIdStr = UUID.randomUUID().toString();
+ UploadLock uploadLock = lockingService.lockUploadByUri("/upload/test/" + uploadIdStr);
- assertThat(
- lockingService.isLocked(new UploadId("000003f1-a850-49de-af03-997272d834c9")), is(true));
+ assertThat(lockingService.isLocked(new UploadId(uploadIdStr)), is(true));
uploadLock.release();
}
@Test
public void isLockedFalse() throws Exception {
- UploadLock uploadLock =
- lockingService.lockUploadByUri("/upload/test/000003f1-a850-49de-af03-997272d834c9");
+ String uploadIdStr = UUID.randomUUID().toString();
+ UploadLock uploadLock = lockingService.lockUploadByUri("/upload/test/" + uploadIdStr);
uploadLock.release();
- assertThat(
- lockingService.isLocked(new UploadId("000003f1-a850-49de-af03-997272d834c9")), is(false));
+ assertThat(lockingService.isLocked(new UploadId(uploadIdStr)), is(false));
}
@Test
@@ -115,8 +121,7 @@ public void lockUploadNotExists() throws Exception {
reset(idFactory);
when(idFactory.readUploadId(nullable(String.class))).thenReturn(null);
- UploadLock uploadLock =
- lockingService.lockUploadByUri("/upload/test/000003f1-a850-49de-af03-997272d834c9");
+ UploadLock uploadLock = lockingService.lockUploadByUri("/upload/test/" + UUID.randomUUID());
assertThat(uploadLock, nullValue());
}
@@ -125,7 +130,7 @@ public void lockUploadNotExists() throws Exception {
public void cleanupStaleLocks() throws Exception {
Path locksPath = storagePath.resolve("locks");
- String activeLock = "000003f1-a850-49de-af03-997272d834c9";
+ String activeLock = UUID.randomUUID().toString();
UploadLock uploadLock = lockingService.lockUploadByUri("/upload/test/" + activeLock);
assertThat(uploadLock, not(nullValue()));
@@ -156,7 +161,8 @@ public void cleanupStaleLocks() throws Exception {
@Test
public void testRegisterAndRequestLockReleaseLocal() throws Exception {
- String uri = "/upload/test/000003f1-a850-49de-af03-997272d834c9";
+ String uploadIdStr = UUID.randomUUID().toString();
+ String uri = "/upload/test/" + uploadIdStr;
byte[] data = new byte[] {1, 2, 3};
ByteArrayInputStream bis = new ByteArrayInputStream(data);
InterruptibleInputStream iis = new InterruptibleInputStream(bis);
@@ -168,15 +174,15 @@ public void testRegisterAndRequestLockReleaseLocal() throws Exception {
assertTrue(iis.isInterrupted());
// Stop file should also be created
- Path stopFilePath =
- storagePath.resolve("locks").resolve("000003f1-a850-49de-af03-997272d834c9.stop");
+ Path stopFilePath = storagePath.resolve("locks").resolve(uploadIdStr + ".stop");
assertTrue(Files.exists(stopFilePath));
Files.deleteIfExists(stopFilePath);
}
@Test
public void testWatchdogInterruptsStreamOnStopFile() throws Exception {
- String uri = "/upload/test/000003f1-a850-49de-af03-997272d834c9";
+ String uploadIdStr = UUID.randomUUID().toString();
+ String uri = "/upload/test/" + uploadIdStr;
byte[] data = new byte[] {1, 2, 3};
ByteArrayInputStream bis = new ByteArrayInputStream(data);
InterruptibleInputStream iis = new InterruptibleInputStream(bis);
@@ -185,15 +191,14 @@ public void testWatchdogInterruptsStreamOnStopFile() throws Exception {
assertFalse(iis.isInterrupted());
// Manually create the stop file (simulating cross-replica signaling)
- Path stopFilePath =
- storagePath.resolve("locks").resolve("000003f1-a850-49de-af03-997272d834c9.stop");
+ Path stopFilePath = storagePath.resolve("locks").resolve(uploadIdStr + ".stop");
Files.createDirectories(stopFilePath.getParent());
Files.write(stopFilePath, new byte[0]);
- // Wait for watchdog to poll (polls every 1000ms, wait up to 2.5s)
+ // Wait for watchdog to poll (polls every 1000ms, wait up to 5s)
long start = System.currentTimeMillis();
- while (!iis.isInterrupted() && System.currentTimeMillis() - start < 2500L) {
- Thread.sleep(100L);
+ while (!iis.isInterrupted() && System.currentTimeMillis() - start < 5000L) {
+ Thread.sleep(50L);
}
assertTrue("Watchdog should have interrupted the stream", iis.isInterrupted());
@@ -202,7 +207,8 @@ public void testWatchdogInterruptsStreamOnStopFile() throws Exception {
@Test
public void testWatchdogTerminatesWhenEmpty() throws Exception {
- String uri = "/upload/test/000003f1-a850-49de-af03-997272d834c9";
+ String uploadIdStr = UUID.randomUUID().toString();
+ String uri = "/upload/test/" + uploadIdStr;
byte[] data = new byte[] {1, 2, 3};
ByteArrayInputStream bis = new ByteArrayInputStream(data);
InterruptibleInputStream iis = new InterruptibleInputStream(bis);
@@ -218,14 +224,13 @@ public void testWatchdogTerminatesWhenEmpty() throws Exception {
// Watchdog should stop (since loop exits after activeLocks is empty)
long start = System.currentTimeMillis();
- while (watchdog.isAlive() && System.currentTimeMillis() - start < 2500L) {
- Thread.sleep(100L);
+ while (watchdog.isAlive() && System.currentTimeMillis() - start < 5000L) {
+ Thread.sleep(50L);
}
assertFalse("Watchdog thread should have terminated", watchdog.isAlive());
// Clean up stop file
- Path stopFilePath =
- storagePath.resolve("locks").resolve("000003f1-a850-49de-af03-997272d834c9.stop");
+ Path stopFilePath = storagePath.resolve("locks").resolve(uploadIdStr + ".stop");
Files.deleteIfExists(stopFilePath);
}
@@ -233,17 +238,16 @@ public void testWatchdogTerminatesWhenEmpty() throws Exception {
public void testDefaultConstructor() throws Exception {
DiskLockingService defaultService = new DiskLockingService(storagePath.toString());
defaultService.setIdFactory(idFactory);
- UploadLock lock =
- defaultService.lockUploadByUri("/upload/test/000003f1-a850-49de-af03-997272d834c9");
+ UploadLock lock = defaultService.lockUploadByUri("/upload/test/" + UUID.randomUUID());
assertThat(lock, not(nullValue()));
lock.close();
}
@Test
public void testRequestLockReleaseIOException() throws Exception {
- String uri = "/upload/test/000003f1-a850-49de-af03-997272d834c9";
- Path stopFilePath =
- storagePath.resolve("locks").resolve("000003f1-a850-49de-af03-997272d834c9.stop");
+ String uploadIdStr = UUID.randomUUID().toString();
+ String uri = "/upload/test/" + uploadIdStr;
+ Path stopFilePath = storagePath.resolve("locks").resolve(uploadIdStr + ".stop");
Files.createDirectories(stopFilePath);
try {
@@ -322,7 +326,8 @@ public void testCleanupStaleLocksWithStaleLockAndStopFile() throws Exception {
@Test
public void testWatchdogRobustnessOnInterruptException() throws Exception {
- String uri = "/upload/test/000003f1-a850-49de-af03-997272d834c9";
+ String uploadIdStr = UUID.randomUUID().toString();
+ String uri = "/upload/test/" + uploadIdStr;
InterruptibleInputStream faultyStream =
new InterruptibleInputStream(new ByteArrayInputStream(new byte[0])) {
@@ -334,14 +339,13 @@ public void interrupt() {
lockingService.registerInputStream(uri, faultyStream);
- Path stopFilePath =
- storagePath.resolve("locks").resolve("000003f1-a850-49de-af03-997272d834c9.stop");
+ Path stopFilePath = storagePath.resolve("locks").resolve(uploadIdStr + ".stop");
Files.createDirectories(stopFilePath.getParent());
Files.write(stopFilePath, new byte[0]);
long start = System.currentTimeMillis();
- while (Files.exists(stopFilePath) && System.currentTimeMillis() - start < 2500L) {
- Thread.sleep(100L);
+ while (Files.exists(stopFilePath) && System.currentTimeMillis() - start < 5000L) {
+ Thread.sleep(50L);
}
Files.deleteIfExists(stopFilePath);
@@ -353,15 +357,16 @@ public void testRequestLockReleaseCreatesParentDirectory() throws Exception {
Path nestedStorage = tempDir.resolve("nested").resolve("sub");
DiskLockingService service = new DiskLockingService(idFactory, nestedStorage.toString());
- UploadId id = new UploadId("000003f1-a850-49de-af03-997272d834c9");
+ String uploadIdStr = UUID.randomUUID().toString();
+ UploadId id = new UploadId(uploadIdStr);
java.lang.reflect.Field urlSafeField = UploadId.class.getDeclaredField("urlSafeValue");
urlSafeField.setAccessible(true);
- urlSafeField.set(id, "subdir/000003f1-a850-49de-af03-997272d834c9");
+ urlSafeField.set(id, "subdir/" + uploadIdStr);
reset(idFactory);
when(idFactory.readUploadId(org.mockito.Mockito.anyString())).thenReturn(id);
- String uri = "/upload/test/subdir/000003f1-a850-49de-af03-997272d834c9";
+ String uri = "/upload/test/subdir/" + uploadIdStr;
service.requestLockRelease(uri);
Path stopFilePath =
@@ -369,16 +374,18 @@ public void testRequestLockReleaseCreatesParentDirectory() throws Exception {
.resolve("locks")
.resolve("subdir")
.resolve("subdir")
- .resolve("000003f1-a850-49de-af03-997272d834c9.stop");
+ .resolve(uploadIdStr + ".stop");
assertTrue(Files.exists(stopFilePath));
Files.deleteIfExists(stopFilePath);
+ service.close();
FileUtils.deleteDirectory(tempDir.toFile());
}
@Test
public void testRequestLockReleaseWithGCedStream() throws Exception {
- String uri = "/upload/test/000003f1-a850-49de-af03-997272d834c9";
+ String uploadIdStr = UUID.randomUUID().toString();
+ String uri = "/upload/test/" + uploadIdStr;
InterruptibleInputStream iis =
new InterruptibleInputStream(new ByteArrayInputStream(new byte[0]));
lockingService.registerInputStream(uri, iis);
@@ -392,17 +399,17 @@ public void testRequestLockReleaseWithGCedStream() throws Exception {
String, java.lang.ref.WeakReference>)
field.get(null);
- java.lang.ref.WeakReference ref =
- map.get("000003f1-a850-49de-af03-997272d834c9");
+ java.lang.ref.WeakReference ref = map.get(uploadIdStr);
ref.clear();
lockingService.requestLockRelease(uri);
- assertFalse(map.containsKey("000003f1-a850-49de-af03-997272d834c9"));
+ assertFalse(map.containsKey(uploadIdStr));
}
@Test
public void testRegisteredLockGetUploadUri() throws Exception {
- String uri = "/upload/test/000003f1-a850-49de-af03-997272d834c9";
+ String uploadIdStr = UUID.randomUUID().toString();
+ String uri = "/upload/test/" + uploadIdStr;
UploadLock lock = lockingService.lockUploadByUri(uri);
org.junit.Assert.assertNotNull(lock);
assertThat(lock.getUploadUri(), is(uri));
@@ -411,7 +418,8 @@ public void testRegisteredLockGetUploadUri() throws Exception {
@Test
public void testWatchdogRemovesClearedWeakReference() throws Exception {
- String uri = "/upload/test/000003f1-a850-49de-af03-997272d834c9";
+ String uploadIdStr = UUID.randomUUID().toString();
+ String uri = "/upload/test/" + uploadIdStr;
InterruptibleInputStream iis =
new InterruptibleInputStream(new ByteArrayInputStream(new byte[0]));
lockingService.registerInputStream(uri, iis);
@@ -425,25 +433,24 @@ public void testWatchdogRemovesClearedWeakReference() throws Exception {
String, java.lang.ref.WeakReference>)
field.get(null);
- java.lang.ref.WeakReference ref =
- map.get("000003f1-a850-49de-af03-997272d834c9");
+ java.lang.ref.WeakReference ref = map.get(uploadIdStr);
org.junit.Assert.assertNotNull(ref);
ref.clear();
long start = System.currentTimeMillis();
- while (map.containsKey("000003f1-a850-49de-af03-997272d834c9")
- && System.currentTimeMillis() - start < 2500L) {
- Thread.sleep(100L);
+ while (map.containsKey(uploadIdStr) && System.currentTimeMillis() - start < 5000L) {
+ Thread.sleep(50L);
}
assertFalse(
"Watchdog should have removed the cleared weak reference from activeLocks",
- map.containsKey("000003f1-a850-49de-af03-997272d834c9"));
+ map.containsKey(uploadIdStr));
}
@Test
public void testWatchdogInterrupted() throws Exception {
- String uri = "/upload/test/000003f1-a850-49de-af03-997272d834c9";
+ String uploadIdStr = UUID.randomUUID().toString();
+ String uri = "/upload/test/" + uploadIdStr;
InterruptibleInputStream iis =
new InterruptibleInputStream(new ByteArrayInputStream(new byte[0]));
lockingService.registerInputStream(uri, iis);
@@ -455,8 +462,8 @@ public void testWatchdogInterrupted() throws Exception {
watchdog.interrupt();
long start = System.currentTimeMillis();
- while (watchdog.isAlive() && System.currentTimeMillis() - start < 2500L) {
- Thread.sleep(100L);
+ while (watchdog.isAlive() && System.currentTimeMillis() - start < 5000L) {
+ Thread.sleep(50L);
}
assertFalse("Watchdog thread should have terminated on interruption", watchdog.isAlive());
@@ -478,7 +485,8 @@ public void testWatchdogUnexpectedException() throws Exception {
String, java.lang.ref.WeakReference>)
field.get(null);
- String uri = "/upload/test/000003f1-a850-49de-af03-997272d834c9";
+ String uploadIdStr = UUID.randomUUID().toString();
+ String uri = "/upload/test/" + uploadIdStr;
InterruptibleInputStream iis =
new InterruptibleInputStream(new ByteArrayInputStream(new byte[0]));
lockingService.registerInputStream(uri, iis);
@@ -489,11 +497,11 @@ public void testWatchdogUnexpectedException() throws Exception {
java.lang.ref.WeakReference mockRef =
org.mockito.Mockito.mock(java.lang.ref.WeakReference.class);
when(mockRef.get()).thenThrow(new RuntimeException("Simulated exception"));
- map.put("trigger-error", mockRef);
+ map.put("trigger-error-" + UUID.randomUUID(), mockRef);
long start = System.currentTimeMillis();
- while (watchdog.isAlive() && System.currentTimeMillis() - start < 2500L) {
- Thread.sleep(100L);
+ while (watchdog.isAlive() && System.currentTimeMillis() - start < 5000L) {
+ Thread.sleep(50L);
}
map.clear();
@@ -501,12 +509,12 @@ public void testWatchdogUnexpectedException() throws Exception {
@Test
public void testRegisteredLockDeleteStopFileIOException() throws Exception {
- String uri = "/upload/test/000003f1-a850-49de-af03-997272d834c9";
+ String uploadIdStr = UUID.randomUUID().toString();
+ String uri = "/upload/test/" + uploadIdStr;
UploadLock lock = lockingService.lockUploadByUri(uri);
org.junit.Assert.assertNotNull(lock);
- Path stopFilePath =
- storagePath.resolve("locks").resolve("000003f1-a850-49de-af03-997272d834c9.stop");
+ Path stopFilePath = storagePath.resolve("locks").resolve(uploadIdStr + ".stop");
Files.createDirectories(stopFilePath);
Files.createFile(stopFilePath.resolve("dummy"));
@@ -558,12 +566,13 @@ public void cleanupStaleLocksWhenStorageDirectoryNotExists() throws Exception {
assertTrue(Files.exists(nonExistentPath.resolve("locks")));
// Cleanup
+ newLockingService.close();
FileUtils.deleteDirectory(nonExistentPath.toFile());
}
@Test
public void testRequestLockReleaseNullLockPath() throws Exception {
- String uri = "/upload/test/000003f1-a850-49de-af03-997272d834c9";
+ String uri = "/upload/test/" + UUID.randomUUID();
// Mock ID factory to return an ID that will result in a null lock path
// We can just return a null UploadId to get null from getPathInStorageDirectory
@@ -654,4 +663,12 @@ public void testStopPathCreationExceptions() throws Exception {
tempDir.toFile().setWritable(true);
FileUtils.deleteDirectory(tempDir.toFile());
}
+
+ @Test
+ public void testClose() throws Exception {
+ DiskLockingService service = new DiskLockingService(storagePath.toString());
+ service.close();
+ // Subsequent close should be idempotent no-op
+ service.close();
+ }
}
diff --git a/src/test/java/me/desair/tus/server/upload/s3/S3ConcatenationServiceTest.java b/src/test/java/me/desair/tus/server/upload/s3/S3ConcatenationServiceTest.java
index e87fce8..41a480b 100644
--- a/src/test/java/me/desair/tus/server/upload/s3/S3ConcatenationServiceTest.java
+++ b/src/test/java/me/desair/tus/server/upload/s3/S3ConcatenationServiceTest.java
@@ -257,6 +257,16 @@ public void testGetConcatenatedBytesNull() throws Exception {
assertNull(concatenationService.getConcatenatedBytes(null));
}
+ @Test
+ public void testGetConcatenatedBytesInProgressReturnsEmptyStream() throws Exception {
+ UploadInfo info = new UploadInfo();
+ info.setId(new UploadId("concat-in-progress"));
+
+ InputStream is = concatenationService.getConcatenatedBytes(info);
+ assertNotNull(is);
+ assertEquals(0, is.available());
+ }
+
/**
* Tests that merging fails and throws an UploadNotFoundException when the child upload has a
* different owner key than the parent/final upload in S3 storage.
diff --git a/src/test/java/me/desair/tus/server/upload/s3/S3LockingServiceTest.java b/src/test/java/me/desair/tus/server/upload/s3/S3LockingServiceTest.java
index 2f0c1e4..0435f3f 100644
--- a/src/test/java/me/desair/tus/server/upload/s3/S3LockingServiceTest.java
+++ b/src/test/java/me/desair/tus/server/upload/s3/S3LockingServiceTest.java
@@ -269,4 +269,181 @@ public void testSanitizePrefixNullOrEmpty() throws Exception {
new S3LockingService(minioClient, "test-bucket", null, 30000L, 0L);
assertNotNull(serviceWithNullPrefix);
}
+
+ @Test
+ public void testClose() throws Exception {
+ lockingService.close();
+ }
+
+ @Test
+ public void testCheckStopSignalForEntryExceptionAndNullId() throws Exception {
+ lockingService.setIdFactory(new me.desair.tus.server.upload.TimeBasedUploadIdFactory());
+ io.minio.MinioClient mockClient = Mockito.mock(io.minio.MinioClient.class);
+ Mockito.when(mockClient.statObject(Mockito.any(io.minio.StatObjectArgs.class)))
+ .thenThrow(new RuntimeException("General S3 Exception"));
+ Mockito.doThrow(new RuntimeException("Remove object failed"))
+ .when(mockClient)
+ .removeObject(Mockito.any(io.minio.RemoveObjectArgs.class));
+
+ S3LockingService service = new S3LockingService(mockClient, "test-bucket");
+ me.desair.tus.server.upload.TimeBasedUploadIdFactory idFactory =
+ new me.desair.tus.server.upload.TimeBasedUploadIdFactory();
+ idFactory.setUploadUri("/files/upload");
+ service.setIdFactory(idFactory);
+
+ ByteArrayInputStream bais = new ByteArrayInputStream("test".getBytes());
+ InterruptibleInputStream stream = new InterruptibleInputStream(bais);
+
+ service.registerInputStream("/files/upload/12345", stream);
+ // Triggers checkStopSignalForEntry & deleteObjectQuietly which catch RuntimeException
+ service.requestLockRelease("/files/upload/12345");
+ }
+
+ @Test
+ public void testCheckStopSignalForEntryHappyPath() throws Exception {
+ MinioClient mockClient = Mockito.mock(MinioClient.class);
+ io.minio.StatObjectResponse mockStat = Mockito.mock(io.minio.StatObjectResponse.class);
+ Mockito.when(mockClient.statObject(Mockito.any(StatObjectArgs.class))).thenReturn(mockStat);
+
+ S3LockingService service =
+ new S3LockingService(mockClient, "test-bucket", "locks", 30000L, 50L);
+ me.desair.tus.server.upload.TimeBasedUploadIdFactory idFactory =
+ new me.desair.tus.server.upload.TimeBasedUploadIdFactory();
+ idFactory.setUploadUri("/files/upload");
+ service.setIdFactory(idFactory);
+
+ ByteArrayInputStream bais = new ByteArrayInputStream("test".getBytes());
+ InterruptibleInputStream stream = new InterruptibleInputStream(bais);
+
+ service.registerInputStream("/files/upload/12345", stream);
+
+ // Wait for background watchdog thread to execute checkStopSignals()
+ long deadline = System.currentTimeMillis() + 2000L;
+ while (!stream.isInterrupted() && System.currentTimeMillis() < deadline) {
+ Thread.sleep(50L);
+ }
+
+ assertTrue(stream.isInterrupted());
+ Mockito.verify(mockClient, Mockito.atLeastOnce()).statObject(Mockito.any(StatObjectArgs.class));
+
+ service.close();
+ }
+
+ @Test
+ public void testCheckStopSignalForEntryNoSuchKey() throws Exception {
+ MinioClient mockClient = Mockito.mock(MinioClient.class);
+ ErrorResponse errorResponse = Mockito.mock(ErrorResponse.class);
+ Mockito.when(errorResponse.code()).thenReturn("NoSuchKey");
+ ErrorResponseException noSuchKeyException =
+ new ErrorResponseException(errorResponse, null, "NoSuchKey");
+
+ Mockito.when(mockClient.statObject(Mockito.any(StatObjectArgs.class)))
+ .thenThrow(noSuchKeyException);
+
+ S3LockingService service =
+ new S3LockingService(mockClient, "test-bucket", "locks", 30000L, 50L);
+ me.desair.tus.server.upload.TimeBasedUploadIdFactory idFactory =
+ new me.desair.tus.server.upload.TimeBasedUploadIdFactory();
+ idFactory.setUploadUri("/files/upload");
+ service.setIdFactory(idFactory);
+
+ ByteArrayInputStream bais = new ByteArrayInputStream("test".getBytes());
+ InterruptibleInputStream stream = new InterruptibleInputStream(bais);
+
+ service.registerInputStream("/files/upload/12345", stream);
+
+ // Wait for background watchdog thread to run checkStopSignals()
+ Thread.sleep(200L);
+
+ Mockito.verify(mockClient, Mockito.atLeastOnce()).statObject(Mockito.any(StatObjectArgs.class));
+ assertFalse(stream.isInterrupted());
+
+ service.close();
+ }
+
+ @Test
+ public void testCleanupStaleLocksWithExpiredAndNonExpiredLocks() throws Exception {
+ MinioClient mockClient = Mockito.mock(MinioClient.class);
+ Item expiredItem = Mockito.mock(Item.class);
+ Mockito.when(expiredItem.objectName()).thenReturn("locks/expired.lock");
+
+ Item validItem = Mockito.mock(Item.class);
+ Mockito.when(validItem.objectName()).thenReturn("locks/valid.lock");
+
+ Result- res1 = new Result
- (expiredItem);
+ Result
- res2 = new Result
- (validItem);
+
+ Mockito.when(mockClient.listObjects(Mockito.any(ListObjectsArgs.class)))
+ .thenReturn(java.util.Arrays.asList(res1, res2));
+
+ S3UploadLock expiredLock =
+ new S3UploadLock(
+ "holder1",
+ "/files/upload/1",
+ "test-bucket",
+ "locks/expired.lock",
+ "locks/expired.stop",
+ 30000L,
+ 1000L);
+ S3UploadLock validLock =
+ new S3UploadLock(
+ "holder2",
+ "/files/upload/2",
+ "test-bucket",
+ "locks/valid.lock",
+ "locks/valid.stop",
+ 30000L,
+ System.currentTimeMillis() + 1000000L);
+
+ GetObjectResponse expiredStream =
+ new GetObjectResponse(
+ null,
+ "test-bucket",
+ "us-east-1",
+ "locks/expired.lock",
+ new ByteArrayInputStream(
+ me.desair.tus.server.util.S3UploadLockJsonSerializer.serialize(expiredLock)
+ .getBytes(StandardCharsets.UTF_8)));
+ GetObjectResponse validStream =
+ new GetObjectResponse(
+ null,
+ "test-bucket",
+ "us-east-1",
+ "locks/valid.lock",
+ new ByteArrayInputStream(
+ me.desair.tus.server.util.S3UploadLockJsonSerializer.serialize(validLock)
+ .getBytes(StandardCharsets.UTF_8)));
+
+ Mockito.when(mockClient.getObject(Mockito.any(GetObjectArgs.class)))
+ .thenReturn(expiredStream)
+ .thenReturn(validStream);
+
+ S3LockingService service = new S3LockingService(mockClient, "test-bucket");
+ service.cleanupStaleLocks();
+
+ // Expired lock object is deleted
+ Mockito.verify(mockClient).removeObject(Mockito.any(io.minio.RemoveObjectArgs.class));
+ }
+
+ @Test
+ public void testRegisterInputStreamNullChecks() {
+ lockingService.registerInputStream(null, Mockito.mock(InputStream.class));
+ lockingService.registerInputStream("/files/upload/123", null);
+ }
+
+ @Test
+ public void testWriteStopSignalException() throws Exception {
+ MinioClient mockClient = Mockito.mock(MinioClient.class);
+ Mockito.when(mockClient.putObject(Mockito.any(PutObjectArgs.class)))
+ .thenThrow(new RuntimeException("S3 Put error"));
+
+ S3LockingService service = new S3LockingService(mockClient, "test-bucket");
+ me.desair.tus.server.upload.TimeBasedUploadIdFactory idFactory =
+ new me.desair.tus.server.upload.TimeBasedUploadIdFactory();
+ idFactory.setUploadUri("/files/upload");
+ service.setIdFactory(idFactory);
+
+ // requestLockRelease calls writeStopSignal which catches RuntimeException
+ service.requestLockRelease("/files/upload/12345");
+ }
}
diff --git a/src/test/java/me/desair/tus/server/upload/s3/S3StorageServiceTest.java b/src/test/java/me/desair/tus/server/upload/s3/S3StorageServiceTest.java
index f0fa45e..2de3d8d 100644
--- a/src/test/java/me/desair/tus/server/upload/s3/S3StorageServiceTest.java
+++ b/src/test/java/me/desair/tus/server/upload/s3/S3StorageServiceTest.java
@@ -255,6 +255,17 @@ public void testGetUploadInfoThrowsIOExceptionOnGenericException() throws Except
storageService.getUploadInfo(new UploadId("24249a5b-01a4-4bf8-b67a-364273bb5a2e"));
}
+ @Test
+ public void testGetS3ObjectKeyWithDuplicatesUploadId() {
+ UploadInfo info = new UploadInfo();
+ UploadId childId = new UploadId("child-id");
+ UploadId parentId = new UploadId("parent-id");
+ info.setId(childId);
+ info.setDuplicatesUploadId(parentId);
+
+ assertEquals("uploads/parent-id", storageService.getS3ObjectKey(info));
+ }
+
@Test(expected = IOException.class)
public void testGetUploadInfoThrowsIOExceptionOnErrorResponseNon404() throws Exception {
ErrorResponse errorResponse = mock(ErrorResponse.class);
diff --git a/src/test/java/me/desair/tus/server/upload/s3/S3UtilsTest.java b/src/test/java/me/desair/tus/server/upload/s3/S3UtilsTest.java
index 1926214..53ee607 100644
--- a/src/test/java/me/desair/tus/server/upload/s3/S3UtilsTest.java
+++ b/src/test/java/me/desair/tus/server/upload/s3/S3UtilsTest.java
@@ -20,6 +20,9 @@ public void testParseErrorResponseCodes() throws Exception {
assertEquals(
S3ErrorType.NO_SUCH_KEY,
S3Utils.parseErrorResponse(createExceptionWithCode("NoSuchBucket")));
+ assertEquals(
+ S3ErrorType.NO_SUCH_KEY,
+ S3Utils.parseErrorResponse(createExceptionWithCode("NoSuchUpload")));
assertEquals(
S3ErrorType.PRECONDITION_FAILED,
S3Utils.parseErrorResponse(createExceptionWithCode("PreconditionFailed")));
@@ -29,6 +32,9 @@ public void testParseErrorResponseCodes() throws Exception {
assertEquals(
S3ErrorType.ACCESS_DENIED,
S3Utils.parseErrorResponse(createExceptionWithCode("AccessDenied")));
+ assertEquals(
+ S3ErrorType.API_NOT_IMPLEMENTED,
+ S3Utils.parseErrorResponse(createExceptionWithCode("APINotImplemented")));
assertEquals(
S3ErrorType.UNKNOWN, S3Utils.parseErrorResponse(createExceptionWithCode("InternalError")));
}
diff --git a/src/test/java/me/desair/tus/server/util/UtilsTest.java b/src/test/java/me/desair/tus/server/util/UtilsTest.java
index bbd4c68..5dbce7a 100644
--- a/src/test/java/me/desair/tus/server/util/UtilsTest.java
+++ b/src/test/java/me/desair/tus/server/util/UtilsTest.java
@@ -512,6 +512,97 @@ public void testIsExistingUploadResourceNullRequestUri() throws Exception {
assertThat(Utils.isExistingUploadResource(request, storageService, "owner"), is(false));
}
+ @Test
+ public void testCreateScheduledDaemonExecutorAndScheduleWatchdog() throws Exception {
+ java.util.concurrent.atomic.AtomicBoolean executed =
+ new java.util.concurrent.atomic.AtomicBoolean(false);
+ java.util.concurrent.ScheduledExecutorService executor =
+ Utils.scheduleWatchdog(
+ "test-watchdog",
+ () -> executed.set(true),
+ 10,
+ 10,
+ java.util.concurrent.TimeUnit.MILLISECONDS);
+
+ assertThat(executor, is(notNullValue()));
+ assertThat(executor.isShutdown(), is(false));
+
+ Thread.sleep(50);
+ assertThat(executed.get(), is(true));
+
+ Utils.shutdownExecutor(executor);
+ assertThat(executor.isShutdown(), is(true));
+ }
+
+ @Test
+ public void testShutdownExecutorNullOrShutdown() {
+ Utils.shutdownExecutor(null);
+
+ java.util.concurrent.ScheduledExecutorService executor =
+ Utils.createScheduledDaemonExecutor("test-shutdown");
+ Utils.shutdownExecutor(executor);
+ assertThat(executor.isShutdown(), is(true));
+
+ Utils.shutdownExecutor(executor);
+ assertThat(executor.isShutdown(), is(true));
+ }
+
+ @Test
+ public void testScheduleWatchdogWithZeroPeriodOrNullTask() {
+ java.util.concurrent.ScheduledExecutorService executor1 =
+ Utils.scheduleWatchdog(
+ "test-zero-period", () -> {}, 0, 0, java.util.concurrent.TimeUnit.SECONDS);
+ assertThat(executor1, is(notNullValue()));
+ Utils.shutdownExecutor(executor1);
+
+ java.util.concurrent.ScheduledExecutorService executor2 =
+ Utils.scheduleWatchdog(
+ "test-null-task", null, 10, 10, java.util.concurrent.TimeUnit.SECONDS);
+ assertThat(executor2, is(notNullValue()));
+ Utils.shutdownExecutor(executor2);
+ }
+
+ @Test
+ public void testShutdownExecutorWithException() {
+ java.util.concurrent.ScheduledExecutorService mockExecutor =
+ mock(java.util.concurrent.ScheduledExecutorService.class);
+ when(mockExecutor.isShutdown()).thenReturn(false);
+ when(mockExecutor.shutdownNow()).thenThrow(new RuntimeException("Shutdown error"));
+
+ Utils.shutdownExecutor(mockExecutor);
+ }
+
+ @Test
+ public void testInterruptStreamNull() {
+ Utils.interruptStream(null);
+ }
+
+ @Test
+ public void testInterruptStreamStandardInputStream() throws Exception {
+ java.io.ByteArrayInputStream bis = new java.io.ByteArrayInputStream(new byte[0]);
+ Utils.interruptStream(bis);
+ }
+
+ @Test
+ public void testInterruptStreamInterruptibleInputStream() {
+ InterruptibleInputStream iis =
+ new InterruptibleInputStream(new java.io.ByteArrayInputStream(new byte[0]));
+ Utils.interruptStream(iis);
+ assertThat(iis.isInterrupted(), is(true));
+ }
+
+ @Test
+ public void testInterruptStreamWithException() {
+ InterruptibleInputStream faultyStream =
+ new InterruptibleInputStream(new java.io.ByteArrayInputStream(new byte[0])) {
+ @Override
+ public void interrupt() {
+ throw new RuntimeException("Error during interrupt");
+ }
+ };
+ Utils.interruptStream(faultyStream);
+ }
+
/** Simple serializable class for testing. */
public static class TestSerializable implements Serializable {
private static final long serialVersionUID = 1L;