30762898309

Committed 02 Aug 2026 07:13PM UTC coverage: 95.854% (-0.6%) from 96.487%

Build # 30762898309

Build Type

Pull #111

github

Committed by

web-flow

Commit Message

Merge 006364903 into 94d541272

Pull Request Pull Request #111: feat: add S3 storage, locking, concatenation, and JSON metadata serialization

Coverage Stats

1714 of 1894 branches covered (90.5%)

Branch coverage included in aggregate %.

895 of 910 new or added lines in 14 files covered. (98.35%)

3835 of 3895 relevant lines covered (98.46%)

6.85 hits per line

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

94.55

/src/main/java/me/desair/tus/server/upload/s3/S3LockingService.java

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

import com.fasterxml.jackson.databind.ObjectMapper;
import io.minio.GetObjectArgs;
import io.minio.ListObjectsArgs;
import io.minio.MinioClient;
import io.minio.PutObjectArgs;
import io.minio.RemoveObjectArgs;
import io.minio.Result;
import io.minio.StatObjectArgs;
import io.minio.errors.ErrorResponseException;
import io.minio.messages.Item;
import java.io.ByteArrayInputStream;
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;
import me.desair.tus.server.exception.UploadAlreadyLockedException;
import me.desair.tus.server.upload.UploadId;
import me.desair.tus.server.upload.UploadIdFactory;
import me.desair.tus.server.upload.UploadLock;
import me.desair.tus.server.upload.UploadLockingService;
import me.desair.tus.server.upload.UuidUploadIdFactory;
import me.desair.tus.server.util.InterruptibleInputStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Distributed S3-backed implementation of {@link UploadLockingService} using the MinIO Java SDK.
 *
 * <p>Key Architecture Features & S3/MinIO Developer Guide:
 *
 * <ul>
 *   <li><b>Distributed Lock Lease Objects</b>: Locks are represented as small JSON lease objects
 *       written to S3 under {@code <locksPrefix>/<UploadId>.lock}. Each lease object records a
 *       unique {@code holderId} and an absolute timestamp {@code expiresAt}.
 *   <li><b>Atomic Lock Acquisition</b>: When an upload request arrives, the server checks whether
 *       an unexpired lock object already exists in S3. If no active lock is found, a new lock
 *       object is written to S3, granting exclusive ownership to the current thread/pod without
 *       requiring Redis or an external database.
 *   <li><b>Heartbeat & Lease Auto-Renewal</b>: Managed locks spawn background daemon threads that
 *       periodically update the lock object in S3, keeping the lease active while long uploads run.
 *   <li><b>Cross-Pod Lock Contention & Interrupt Signals</b>: When a concurrent request arrives for
 *       a locked upload (e.g., HEAD or DELETE while a PATCH is streaming data on another pod), the
 *       service writes a {@code <locksPrefix>/<UploadId>.stop} signal object to S3. A background
 *       watchdog thread on the pod holding the lock detects the {@code .stop} file and interrupts
 *       the active input stream immediately, resolving lock contention cleanly across Kubernetes
 *       pods.
 * </ul>
 */
public class S3LockingService implements UploadLockingService {

  private static final Logger log = LoggerFactory.getLogger(S3LockingService.class);
  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

  public static final String DEFAULT_LOCKS_PREFIX = "locks/";
  public static final long DEFAULT_LEASE_DURATION_MS = 30_000L; // 30 seconds
  public static final long DEFAULT_POLL_INTERVAL_MS = 2_000L; // 2 seconds

  private final MinioClient minioClient;
  private final String bucket;
  private final String locksPrefix;
  private final long leaseDurationMs;
  private final long pollIntervalMs;

  private UploadIdFactory idFactory = new UuidUploadIdFactory();
  private final Map<String, InputStream> activeInputStreams = new ConcurrentHashMap<>();
  private final ScheduledExecutorService watchdogExecutor;

  /**
   * Basic constructor using default lock prefix ("locks/"), 30s lease duration, and 2s polling
   * interval.
   *
   * @param minioClient Pre-configured MinIO Client
   * @param bucket Target S3 bucket name
   */
  public S3LockingService(MinioClient minioClient, String bucket) {
    this(
        minioClient,
        bucket,
        DEFAULT_LOCKS_PREFIX,
        DEFAULT_LEASE_DURATION_MS,
        DEFAULT_POLL_INTERVAL_MS);
  }

  /**
   * Full constructor allowing custom configuration for all locking parameters.
   *
   * @param minioClient Pre-configured MinIO Client
   * @param bucket Target S3 bucket name
   * @param locksPrefix Object key prefix for locks and stop signals
   * @param leaseDurationMs Lock lease duration in milliseconds
   * @param pollIntervalMs Watchdog poll interval for lock contention interrupt signals
   */
  public S3LockingService(
      MinioClient minioClient,
      String bucket,
      String locksPrefix,
      long leaseDurationMs,
      long pollIntervalMs) {
    this.minioClient = Objects.requireNonNull(minioClient, "MinioClient must not be null");
    this.bucket = Objects.requireNonNull(bucket, "Bucket must not be null");
    this.locksPrefix = sanitizePrefix(locksPrefix);
    this.leaseDurationMs = leaseDurationMs;
    this.pollIntervalMs = pollIntervalMs;

    // 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);
    }
  }

  @Override
  public UploadLock lockUploadByUri(String requestUri) throws TusException, IOException {
    UploadId uploadId = idFactory.readUploadId(requestUri);
    if (uploadId == null) {
      return null;
    }

    String lockKey = buildLockKey(uploadId);
    String stopKey = buildStopKey(uploadId);
    String holderId = UUID.randomUUID().toString();

    // Attempt lock acquisition or clear expired lock
    boolean acquired = acquireOrEvictExpiredLock(lockKey, holderId);
    if (!acquired) {
      throw new UploadAlreadyLockedException("Upload " + uploadId + " is currently locked");
    }

    // Create and return S3UploadLock instance with heartbeat lease auto-renewal
    return new S3UploadLock(
        minioClient,
        bucket,
        lockKey,
        stopKey,
        holderId,
        leaseDurationMs,
        requestUri,
        activeInputStreams);
  }

  @Override
  public void cleanupStaleLocks() throws IOException {
    try {
      // List all object keys under locksPrefix in S3
      Iterable<Result<Item>> results =
          minioClient.listObjects(
              ListObjectsArgs.builder().bucket(bucket).prefix(locksPrefix).build());

      for (Result<Item> result : results) {
        Item item = result.get();
        // Remove expired .lock lease objects
        if (item.objectName().endsWith(".lock") && isLockExpired(item.objectName())) {
          deleteObjectQuietly(item.objectName());
        }
      }
    } catch (Exception e) {
      throw new IOException("Failed to cleanup stale S3 locks", e);
    }
  }

  @Override
  public boolean isLocked(UploadId id) {
    if (id == null) {
      return false;
    }
    String lockKey = buildLockKey(id);
    return !isLockExpired(lockKey);
  }

  @Override
  public void setIdFactory(UploadIdFactory idFactory) {
    if (idFactory != null) {
      this.idFactory = idFactory;
    }
  }

  @Override
  public void registerInputStream(String requestUri, InputStream inputStream) {
    if (requestUri != null && inputStream != null) {
      activeInputStreams.put(requestUri, inputStream);
    }
  }

  @Override
  public void requestLockRelease(String requestUri) {
    if (requestUri == null) {
      return;
    }

    // Step 1: Interrupt local active payload byte stream if hosted on this node
    InputStream activeStream = activeInputStreams.get(requestUri);
    if (activeStream != null) {
      interruptStream(activeStream);
    }

    // Step 2: Write a .stop signal object to S3 to signal lock contention across remote nodes/pods
    UploadId uploadId = idFactory.readUploadId(requestUri);
    if (uploadId != null) {
      writeStopSignal(uploadId);
    }
  }

  // HELPER METHODS & LOCK MANAGEMENT LOGIC

  private boolean acquireOrEvictExpiredLock(String lockKey, String holderId) {
    boolean acquired = attemptLockAcquisition(lockKey, holderId);
    if (!acquired && isLockExpired(lockKey)) {
      deleteObjectQuietly(lockKey);
      acquired = attemptLockAcquisition(lockKey, holderId);
    }
    return acquired;
  }

  private boolean attemptLockAcquisition(String lockKey, String holderId) {
    if (!isLockExpired(lockKey)) {
      return false;
    }

    long expiresAt = System.currentTimeMillis() + leaseDurationMs;
    try {
      byte[] lockContentBytes = OBJECT_MAPPER.writeValueAsBytes(new LockData(holderId, expiresAt));

      // Put lock lease object to S3
      minioClient.putObject(
          PutObjectArgs.builder().bucket(bucket).object(lockKey).stream(
                  new ByteArrayInputStream(lockContentBytes), (long) lockContentBytes.length, -1L)
              .build());
      return true;
    } catch (Exception e) {
      log.warn("Unexpected error acquiring S3 lock for key {}", lockKey, e);
      return false;
    }
  }

  private boolean isLockExpired(String lockKey) {
    try (InputStream stream =
        minioClient.getObject(GetObjectArgs.builder().bucket(bucket).object(lockKey).build())) {

      LockData lockData = OBJECT_MAPPER.readValue(stream, LockData.class);
      return lockData.expiresAt < System.currentTimeMillis();
    } catch (ErrorResponseException e) {
      if (S3Utils.parseErrorResponse(e) == S3ErrorType.NO_SUCH_KEY) {
        return true; // Key missing -> Not locked
      }
      return true;
    } catch (Exception e) {
      log.debug("Failed to read lock object {}, treating as expired", lockKey, e);
      return true;
    }
  }

  private void writeStopSignal(UploadId uploadId) {
    String stopKey = buildStopKey(uploadId);
    try {
      byte[] empty = new byte[0];
      // Write empty .stop signal object to S3
      minioClient.putObject(
          PutObjectArgs.builder().bucket(bucket).object(stopKey).stream(
                  new ByteArrayInputStream(empty), 0L, -1L)
              .build());
    } catch (Exception e) {
      log.debug("Failed to write lock stop signal to S3 key {}", stopKey, e);
    }
  }

  private void checkStopSignals() {
    for (Map.Entry<String, InputStream> entry : activeInputStreams.entrySet()) {
      checkStopSignalForEntry(entry.getKey(), entry.getValue());
    }
  }

  private void checkStopSignalForEntry(String uri, InputStream inputStream) {
    UploadId uploadId = idFactory.readUploadId(uri);
    if (uploadId == null) {
      return;
    }

    String stopKey = buildStopKey(uploadId);
    try {
      // Check if a .stop signal object was written by another pod requesting lock release
      minioClient.statObject(StatObjectArgs.builder().bucket(bucket).object(stopKey).build());
      // Remote stop signal object found! Interrupt local byte stream immediately
      interruptStream(inputStream);
    } catch (ErrorResponseException e) {
      if ("NoSuchKey".equalsIgnoreCase(e.errorResponse().code())) {
        // Normal state: no stop signal object in S3
        return;
      }
    } catch (Exception e) {
      log.debug("Error checking stop signal for {}", stopKey, e);
    }
  }

  private void interruptStream(InputStream is) {
    if (is instanceof InterruptibleInputStream) {
      ((InterruptibleInputStream) is).interrupt();
    } else {
      try {
        is.close();
      } catch (Exception ignored) {
      }
    }
  }

  private void deleteObjectQuietly(String key) {
    try {
      minioClient.removeObject(RemoveObjectArgs.builder().bucket(bucket).object(key).build());
    } catch (Exception e) {
      log.debug("Failed to delete S3 object key {}", key, e);
    }
  }

  private String sanitizePrefix(String prefix) {
    if (prefix == null || prefix.isEmpty()) {
      return "";
    }
    String result = prefix.startsWith("/") ? prefix.substring(1) : prefix;
    return result.endsWith("/") ? result : result + "/";
  }

  private String buildLockKey(UploadId uploadId) {
    return locksPrefix + uploadId.toString() + ".lock";
  }

  private String buildStopKey(UploadId uploadId) {
    return locksPrefix + uploadId.toString() + ".stop";
  }

  public static class LockData {
    private String holderId;
    private long expiresAt;
    private long acquiredAt;

    public LockData() {}

    public LockData(String holderId, long expiresAt) {
      this.holderId = holderId;
      this.expiresAt = expiresAt;
      this.acquiredAt = System.currentTimeMillis();
    }

    public String getHolderId() {
      return holderId;
    }

    public void setHolderId(String holderId) {
      this.holderId = holderId;
    }

    public long getExpiresAt() {
      return expiresAt;
    }

    public void setExpiresAt(long expiresAt) {
      this.expiresAt = expiresAt;
    }

    public long getAcquiredAt() {
      return acquiredAt;
    }

    public void setAcquiredAt(long acquiredAt) {
      this.acquiredAt = acquiredAt;
    }
  }
}

1	package me.desair.tus.server.upload.s3;
2
3	import com.fasterxml.jackson.databind.ObjectMapper;
4	import io.minio.GetObjectArgs;
5	import io.minio.ListObjectsArgs;
6	import io.minio.MinioClient;
7	import io.minio.PutObjectArgs;
8	import io.minio.RemoveObjectArgs;
9	import io.minio.Result;
10	import io.minio.StatObjectArgs;
11	import io.minio.errors.ErrorResponseException;
12	import io.minio.messages.Item;
13	import java.io.ByteArrayInputStream;
14	import java.io.IOException;
15	import java.io.InputStream;
16	import java.util.Map;
17	import java.util.Objects;
18	import java.util.UUID;
19	import java.util.concurrent.ConcurrentHashMap;
20	import java.util.concurrent.Executors;
21	import java.util.concurrent.ScheduledExecutorService;
22	import java.util.concurrent.TimeUnit;
23	import me.desair.tus.server.exception.TusException;
24	import me.desair.tus.server.exception.UploadAlreadyLockedException;
25	import me.desair.tus.server.upload.UploadId;
26	import me.desair.tus.server.upload.UploadIdFactory;
27	import me.desair.tus.server.upload.UploadLock;
28	import me.desair.tus.server.upload.UploadLockingService;
29	import me.desair.tus.server.upload.UuidUploadIdFactory;
30	import me.desair.tus.server.util.InterruptibleInputStream;
31	import org.slf4j.Logger;
32	import org.slf4j.LoggerFactory;
33
34	/**
35	* Distributed S3-backed implementation of {@link UploadLockingService} using the MinIO Java SDK.
36	*
37	* <p>Key Architecture Features & S3/MinIO Developer Guide:
38	*
39	* <ul>
40	* <li><b>Distributed Lock Lease Objects</b>: Locks are represented as small JSON lease objects
41	* written to S3 under {@code <locksPrefix>/<UploadId>.lock}. Each lease object records a
42	* unique {@code holderId} and an absolute timestamp {@code expiresAt}.
43	* <li><b>Atomic Lock Acquisition</b>: When an upload request arrives, the server checks whether
44	* an unexpired lock object already exists in S3. If no active lock is found, a new lock
45	* object is written to S3, granting exclusive ownership to the current thread/pod without
46	* requiring Redis or an external database.
47	* <li><b>Heartbeat & Lease Auto-Renewal</b>: Managed locks spawn background daemon threads that
48	* periodically update the lock object in S3, keeping the lease active while long uploads run.
49	* <li><b>Cross-Pod Lock Contention & Interrupt Signals</b>: When a concurrent request arrives for
50	* a locked upload (e.g., HEAD or DELETE while a PATCH is streaming data on another pod), the
51	* service writes a {@code <locksPrefix>/<UploadId>.stop} signal object to S3. A background
52	* watchdog thread on the pod holding the lock detects the {@code .stop} file and interrupts
53	* the active input stream immediately, resolving lock contention cleanly across Kubernetes
54	* pods.
55	* </ul>
56	*/
57	public class S3LockingService implements UploadLockingService {
58
59	private static final Logger log = LoggerFactory.getLogger(S3LockingService.class);	6✔
60	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();	10✔
61
62	public static final String DEFAULT_LOCKS_PREFIX = "locks/";
63	public static final long DEFAULT_LEASE_DURATION_MS = 30_000L; // 30 seconds
64	public static final long DEFAULT_POLL_INTERVAL_MS = 2_000L; // 2 seconds
65
66	private final MinioClient minioClient;
67	private final String bucket;
68	private final String locksPrefix;
69	private final long leaseDurationMs;
70	private final long pollIntervalMs;
71
72	private UploadIdFactory idFactory = new UuidUploadIdFactory();	10✔
73	private final Map<String, InputStream> activeInputStreams = new ConcurrentHashMap<>();	10✔
74	private final ScheduledExecutorService watchdogExecutor;
75
76	/**
77	* Basic constructor using default lock prefix ("locks/"), 30s lease duration, and 2s polling
78	* interval.
79	*
80	* @param minioClient Pre-configured MinIO Client
81	* @param bucket Target S3 bucket name
82	*/
83	public S3LockingService(MinioClient minioClient, String bucket) {
84	this(	14✔
85	minioClient,
86	bucket,
87	DEFAULT_LOCKS_PREFIX,
88	DEFAULT_LEASE_DURATION_MS,
89	DEFAULT_POLL_INTERVAL_MS);
90	}	2✔
91
92	/**
93	* Full constructor allowing custom configuration for all locking parameters.
94	*
95	* @param minioClient Pre-configured MinIO Client
96	* @param bucket Target S3 bucket name
97	* @param locksPrefix Object key prefix for locks and stop signals
98	* @param leaseDurationMs Lock lease duration in milliseconds
99	* @param pollIntervalMs Watchdog poll interval for lock contention interrupt signals
100	*/
101	public S3LockingService(
102	MinioClient minioClient,
103	String bucket,
104	String locksPrefix,
105	long leaseDurationMs,
106	long pollIntervalMs) {	4✔
107	this.minioClient = Objects.requireNonNull(minioClient, "MinioClient must not be null");	12✔
108	this.bucket = Objects.requireNonNull(bucket, "Bucket must not be null");	12✔
109	this.locksPrefix = sanitizePrefix(locksPrefix);	10✔
110	this.leaseDurationMs = leaseDurationMs;	6✔
111	this.pollIntervalMs = pollIntervalMs;	6✔
112
113	// Background watchdog thread to poll S3 for .stop contention signals across pods
114	this.watchdogExecutor =	4✔
115	Executors.newSingleThreadScheduledExecutor(	4✔
116	r -> {
117	Thread t = new Thread(r, "s3-lock-watchdog");	12✔
118	t.setDaemon(true);	6✔
119	return t;	4✔
120	});
121
122	if (pollIntervalMs > 0) {	8✔
123	this.watchdogExecutor.scheduleAtFixedRate(	18✔
124	this::checkStopSignals, pollIntervalMs, pollIntervalMs, TimeUnit.MILLISECONDS);
125	}
126	}	2✔
127
128	@Override
129	public UploadLock lockUploadByUri(String requestUri) throws TusException, IOException {
130	UploadId uploadId = idFactory.readUploadId(requestUri);	10✔
131	if (uploadId == null) {	4✔
132	return null;	4✔
133	}
134
135	String lockKey = buildLockKey(uploadId);	8✔
136	String stopKey = buildStopKey(uploadId);	8✔
137	String holderId = UUID.randomUUID().toString();	6✔
138
139	// Attempt lock acquisition or clear expired lock
140	boolean acquired = acquireOrEvictExpiredLock(lockKey, holderId);	10✔
141	if (!acquired) {	4✔
142	throw new UploadAlreadyLockedException("Upload " + uploadId + " is currently locked");	14✔
143	}
144
145	// Create and return S3UploadLock instance with heartbeat lease auto-renewal
146	return new S3UploadLock(	32✔
147	minioClient,
148	bucket,
149	lockKey,
150	stopKey,
151	holderId,
152	leaseDurationMs,
153	requestUri,
154	activeInputStreams);
155	}
156
157	@Override
158	public void cleanupStaleLocks() throws IOException {
159	try {
160	// List all object keys under locksPrefix in S3
161	Iterable<Result<Item>> results =	4✔
162	minioClient.listObjects(	4✔
163	ListObjectsArgs.builder().bucket(bucket).prefix(locksPrefix).build());	20✔
164
165	for (Result<Item> result : results) {	16✔
166	Item item = result.get();	4✔
167	// Remove expired .lock lease objects
168	if (item.objectName().endsWith(".lock") && isLockExpired(item.objectName())) {	10!
169	deleteObjectQuietly(item.objectName());	4✔
170	}
171	}	1✔
172	} catch (Exception e) {	1✔
173	throw new IOException("Failed to cleanup stale S3 locks", e);	6✔
174	}	2✔
175	}	2✔
176
177	@Override
178	public boolean isLocked(UploadId id) {
179	if (id == null) {	4✔
180	return false;	2✔
181	}
182	String lockKey = buildLockKey(id);	8✔
183	return !isLockExpired(lockKey);	16✔
184	}
185
186	@Override
187	public void setIdFactory(UploadIdFactory idFactory) {
188	if (idFactory != null) {	4!
189	this.idFactory = idFactory;	6✔
190	}
191	}	2✔
192
193	@Override
194	public void registerInputStream(String requestUri, InputStream inputStream) {
195	if (requestUri != null && inputStream != null) {	8!
196	activeInputStreams.put(requestUri, inputStream);	12✔
197	}
198	}	2✔
199
200	@Override
201	public void requestLockRelease(String requestUri) {
202	if (requestUri == null) {	4✔
203	return;	1✔
204	}
205
206	// Step 1: Interrupt local active payload byte stream if hosted on this node
207	InputStream activeStream = activeInputStreams.get(requestUri);	12✔
208	if (activeStream != null) {	4✔
209	interruptStream(activeStream);	6✔
210	}
211
212	// Step 2: Write a .stop signal object to S3 to signal lock contention across remote nodes/pods
213	UploadId uploadId = idFactory.readUploadId(requestUri);	10✔
214	if (uploadId != null) {	4✔
215	writeStopSignal(uploadId);	6✔
216	}
217	}	2✔
218
219	// HELPER METHODS & LOCK MANAGEMENT LOGIC
220
221	private boolean acquireOrEvictExpiredLock(String lockKey, String holderId) {
222	boolean acquired = attemptLockAcquisition(lockKey, holderId);	10✔
223	if (!acquired && isLockExpired(lockKey)) {	12!
224	deleteObjectQuietly(lockKey);	3✔
225	acquired = attemptLockAcquisition(lockKey, holderId);	5✔
226	}
227	return acquired;	4✔
228	}
229
230	private boolean attemptLockAcquisition(String lockKey, String holderId) {
231	if (!isLockExpired(lockKey)) {	8✔
232	return false;	2✔
233	}
234
235	long expiresAt = System.currentTimeMillis() + leaseDurationMs;	10✔
236	try {
237	byte[] lockContentBytes = OBJECT_MAPPER.writeValueAsBytes(new LockData(holderId, expiresAt));	16✔
238
239	// Put lock lease object to S3
240	minioClient.putObject(	8✔
241	PutObjectArgs.builder().bucket(bucket).object(lockKey).stream(	32✔
242	new ByteArrayInputStream(lockContentBytes), (long) lockContentBytes.length, -1L)	6✔
243	.build());	4✔
244	return true;	4✔
245	} catch (Exception e) {	1✔
246	log.warn("Unexpected error acquiring S3 lock for key {}", lockKey, e);	5✔
247	return false;	2✔
248	}
249	}
250
251	private boolean isLockExpired(String lockKey) {
252	try (InputStream stream =	4✔
253	minioClient.getObject(GetObjectArgs.builder().bucket(bucket).object(lockKey).build())) {	24✔
254
255	LockData lockData = OBJECT_MAPPER.readValue(stream, LockData.class);	12✔
256	return lockData.expiresAt < System.currentTimeMillis();	20✔
257	} catch (ErrorResponseException e) {	2✔
258	if (S3Utils.parseErrorResponse(e) == S3ErrorType.NO_SUCH_KEY) {	8✔
259	return true; // Key missing -> Not locked	4✔
260	}
261	return true;	2✔
262	} catch (Exception e) {	1✔
263	log.debug("Failed to read lock object {}, treating as expired", lockKey, e);	5✔
264	return true;	2✔
265	}
266	}
267
268	private void writeStopSignal(UploadId uploadId) {
269	String stopKey = buildStopKey(uploadId);	8✔
270	try {
271	byte[] empty = new byte[0];	6✔
272	// Write empty .stop signal object to S3
273	minioClient.putObject(	8✔
274	PutObjectArgs.builder().bucket(bucket).object(stopKey).stream(	28✔
275	new ByteArrayInputStream(empty), 0L, -1L)	6✔
276	.build());	4✔
277	} catch (Exception e) {	1✔
278	log.debug("Failed to write lock stop signal to S3 key {}", stopKey, e);	5✔
279	}	2✔
280	}	2✔
281
282	private void checkStopSignals() {
283	for (Map.Entry<String, InputStream> entry : activeInputStreams.entrySet()) {	24✔
284	checkStopSignalForEntry(entry.getKey(), entry.getValue());	16✔
285	}	2✔
286	}	2✔
287
288	private void checkStopSignalForEntry(String uri, InputStream inputStream) {
289	UploadId uploadId = idFactory.readUploadId(uri);	10✔
290	if (uploadId == null) {	4✔
291	return;	1✔
292	}
293
294	String stopKey = buildStopKey(uploadId);	8✔
295	try {
296	// Check if a .stop signal object was written by another pod requesting lock release
297	minioClient.statObject(StatObjectArgs.builder().bucket(bucket).object(stopKey).build());	14✔
298	// Remote stop signal object found! Interrupt local byte stream immediately
299	interruptStream(inputStream);	3✔
300	} catch (ErrorResponseException e) {	1✔
301	if ("NoSuchKey".equalsIgnoreCase(e.errorResponse().code())) {	6!
302	// Normal state: no stop signal object in S3
303	return;	1✔
304	}
305	} catch (Exception e) {	1✔
306	log.debug("Error checking stop signal for {}", stopKey, e);	5✔
307	}	1✔
308	}	2✔
309
310	private void interruptStream(InputStream is) {
311	if (is instanceof InterruptibleInputStream) {	6✔
312	((InterruptibleInputStream) is).interrupt();	8✔
313	} else {
314	try {
315	is.close();	2✔
316	} catch (Exception ignored) {	1✔
317	}	1✔
318	}
319	}	2✔
320
321	private void deleteObjectQuietly(String key) {
322	try {
323	minioClient.removeObject(RemoveObjectArgs.builder().bucket(bucket).object(key).build());	13✔
NEW 324	} catch (Exception e) {	×
NEW 325	log.debug("Failed to delete S3 object key {}", key, e);	×
326	}	1✔
327	}	1✔
328
329	private String sanitizePrefix(String prefix) {
330	if (prefix == null \|\| prefix.isEmpty()) {	10✔
331	return "";	2✔
332	}
333	String result = prefix.startsWith("/") ? prefix.substring(1) : prefix;	12!
334	return result.endsWith("/") ? result : result + "/";	14!
335	}
336
337	private String buildLockKey(UploadId uploadId) {
338	return locksPrefix + uploadId.toString() + ".lock";	12✔
339	}
340
341	private String buildStopKey(UploadId uploadId) {
342	return locksPrefix + uploadId.toString() + ".stop";	12✔
343	}
344
345	public static class LockData {
346	private String holderId;
347	private long expiresAt;
348	private long acquiredAt;
349
350	public LockData() {}	6✔
351
352	public LockData(String holderId, long expiresAt) {	4✔
353	this.holderId = holderId;	6✔
354	this.expiresAt = expiresAt;	6✔
355	this.acquiredAt = System.currentTimeMillis();	6✔
356	}	2✔
357
358	public String getHolderId() {
359	return holderId;	6✔
360	}
361
362	public void setHolderId(String holderId) {
363	this.holderId = holderId;	6✔
364	}	2✔
365
366	public long getExpiresAt() {
367	return expiresAt;	6✔
368	}
369
370	public void setExpiresAt(long expiresAt) {
371	this.expiresAt = expiresAt;	6✔
372	}	2✔
373
374	public long getAcquiredAt() {
375	return acquiredAt;	6✔
376	}
377
378	public void setAcquiredAt(long acquiredAt) {
379	this.acquiredAt = acquiredAt;	3✔
380	}	1✔
381	}
382	}

tomdesair / tus-java-server / 30762898309

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

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