From 341bfb4a0794116da13e0f5dcb4709fea9495d49 Mon Sep 17 00:00:00 2001 From: wyjh Date: Tue, 20 Oct 2026 21:15:01 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E5=A2=9E=E5=BC=BA=20BMCLAPI=20=E4=BD=8E?= =?UTF-8?q?=E9=80=9F=E8=8A=82=E7=82=B9=E9=87=8D=E8=AF=95=E4=B8=8E=E6=96=AD?= =?UTF-8?q?=E7=82=B9=E7=BB=AD=E4=BC=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../jackhuang/hmcl/task/CacheFileTask.java | 7 +- .../org/jackhuang/hmcl/task/FetchTask.java | 115 ++++- .../java/org/jackhuang/hmcl/task/GetTask.java | 9 + .../hmcl/task/BMCLAPIRetryPolicyTest.java | 81 ++++ .../jackhuang/hmcl/task/FetchTaskTest.java | 396 ++++++++++++++++++ 5 files changed, 584 insertions(+), 24 deletions(-) create mode 100644 HMCLCore/src/test/java/org/jackhuang/hmcl/task/BMCLAPIRetryPolicyTest.java create mode 100644 HMCLCore/src/test/java/org/jackhuang/hmcl/task/FetchTaskTest.java diff --git a/HMCLCore/src/main/java/org/jackhuang/hmcl/task/CacheFileTask.java b/HMCLCore/src/main/java/org/jackhuang/hmcl/task/CacheFileTask.java index 45e403b63b2..93a37a28785 100644 --- a/HMCLCore/src/main/java/org/jackhuang/hmcl/task/CacheFileTask.java +++ b/HMCLCore/src/main/java/org/jackhuang/hmcl/task/CacheFileTask.java @@ -146,6 +146,7 @@ protected Context getContext(@Nullable UrlResponseInfo response, boolean checkET throw new IOException("Remote response metadata is unavailable"); } + @Nullable String checksum = expectedSha1 != null ? expectedSha1 : bmclapiHash; return new Context() { private final Path temp = Files.createTempFile("hmcl-download-", null); private final FileChannel fileOutput = FileChannel.open(temp, @@ -183,9 +184,11 @@ public void close() throws IOException { } try { - if (expectedSha1 != null) { + if (checksum != null) { ChecksumMismatchException.verifyChecksum( - temp, CacheRepository.SHA1, expectedSha1); + temp, CacheRepository.SHA1, checksum); + } + if (expectedSha1 != null) { setResult(repository.cacheFile( temp, CacheRepository.SHA1, expectedSha1)); } else { diff --git a/HMCLCore/src/main/java/org/jackhuang/hmcl/task/FetchTask.java b/HMCLCore/src/main/java/org/jackhuang/hmcl/task/FetchTask.java index 418440cb0ba..1215450a9d8 100644 --- a/HMCLCore/src/main/java/org/jackhuang/hmcl/task/FetchTask.java +++ b/HMCLCore/src/main/java/org/jackhuang/hmcl/task/FetchTask.java @@ -30,6 +30,7 @@ import java.io.*; import java.net.HttpURLConnection; import java.net.MalformedURLException; +import java.net.SocketTimeoutException; import java.net.URLConnection; import java.nio.file.Path; import java.time.Duration; @@ -83,6 +84,7 @@ private Context getContext() throws IOException { return getContext(null, false, null); } + /// Creates a sink that uses the BMCLAPI hash unless an explicit integrity check takes precedence. protected abstract Context getContext(@Nullable UrlResponseInfo response, boolean checkETag, @Nullable String bmclapiHash) throws IOException; @Override @@ -142,10 +144,12 @@ public void execute() throws Exception { } } + /// Retains byte ranges and file validators independently of the download sink. private static final class HttpResumeContext { private static final Pattern CONTENT_RANGE_PATTERN = Pattern.compile("bytes ([0-9]+)-([0-9]+)/([0-9]+)"); - static @Nullable FetchTask.HttpResumeContext of(UrlResponseInfo response) throws IOException { + /// Creates resume state for an uncompressed response with a known length and byte-range support. + static @Nullable FetchTask.HttpResumeContext of(UrlResponseInfo response, @Nullable String bmclapiHash) throws IOException { if (response.responseCode() != HttpURLConnection.HTTP_OK) return null; @@ -161,13 +165,13 @@ private static final class HttpResumeContext { if (contentLength < 0) return null; - String eTag = response.headers().firstValue("etag").orElse(null); - String strongETag = isStrongETag(eTag) ? eTag : null; - String lastModified = response.headers().firstValue("last-modified").orElse(null); - if (strongETag == null && StringUtils.isBlank(lastModified)) + @Nullable String eTag = response.headers().firstValue("etag").orElse(null); + @Nullable String strongETag = isStrongETag(eTag) ? eTag : null; + @Nullable String lastModified = response.headers().firstValue("last-modified").orElse(null); + if (bmclapiHash == null && strongETag == null && StringUtils.isBlank(lastModified)) return null; - return new HttpResumeContext(response.url(), contentLength, strongETag, lastModified); + return new HttpResumeContext(response.url(), contentLength, strongETag, lastModified, bmclapiHash); } /// Final response URL whose partial content is retained for resumption. @@ -176,25 +180,34 @@ private static final class HttpResumeContext { private final @Nullable String strongETag; private final @Nullable String lastModified; + /// SHA-1 supplied by the redirect chain, which must match when resuming from another node. + private final @Nullable String bmclapiHash; + long countUncompressed; - /// Records the response URL, total byte length, and available validators for resumption. - private HttpResumeContext(WebURL url, long contentLength, @Nullable String strongETag, @Nullable String lastModified) { + /// Records the response URL, total length, and validators for resumption. + private HttpResumeContext(WebURL url, long contentLength, @Nullable String strongETag, @Nullable String lastModified, + @Nullable String bmclapiHash) { this.url = url; this.contentLength = contentLength; this.strongETag = strongETag; this.lastModified = lastModified; + this.bmclapiHash = bmclapiHash; } private static boolean isStrongETag(@Nullable String eTag) { return StringUtils.isNotBlank(eTag) && !eTag.regionMatches(true, 0, "W/", 0, 2); } - String ifRange() { + /// Uses content hashes for BMCLAPI resumption instead of sending another node's HTTP validator. + @Nullable String ifRange() { + if (bmclapiHash != null) + return null; return strongETag != null ? strongETag : Objects.requireNonNull(lastModified); } - boolean canResume(int statusCode, UrlResponseInfo response) throws IOException { + /// Validates the response range, total length, and file identity before retaining partial content. + boolean canResume(int statusCode, UrlResponseInfo response, @Nullable String bmclapiHash) throws IOException { if (statusCode != HttpURLConnection.HTTP_PARTIAL) return false; @@ -206,8 +219,11 @@ boolean canResume(int statusCode, UrlResponseInfo response) throws IOException { if (this.contentLength != contentLength + this.countUncompressed) return false; - if (strongETag != null) { - String eTag = response.headers().firstValue("etag").orElse(null); + if (this.bmclapiHash != null) { + if (!this.bmclapiHash.equalsIgnoreCase(bmclapiHash)) + return false; + } else if (strongETag != null) { + @Nullable String eTag = response.headers().firstValue("etag").orElse(null); if (!strongETag.equals(eTag)) return false; } else { @@ -243,14 +259,17 @@ boolean hasPartialContent() { } } + /// Reads the response body, retaining resumable fragments and applying the optional node retry policy. private void download(Context context, @Nullable FetchTask.HttpResumeContext resume, InputStream inputStream, long contentLength, - ContentEncoding contentEncoding) throws IOException, InterruptedException { + ContentEncoding contentEncoding, @Nullable BMCLAPIRetryPolicy retryPolicy) throws IOException, InterruptedException { boolean success; try (var counter = new CounterInputStream(inputStream); var input = contentEncoding.wrap(counter)) { long lastDownloaded = 0L; + if (retryPolicy != null) + retryPolicy.startAttempt(System.nanoTime()); byte[] buffer = new byte[IOUtils.DEFAULT_BUFFER_SIZE]; while (true) { if (isCancelled()) break; @@ -275,6 +294,10 @@ private void download(Context context, updateDownloadSpeed(counter.downloaded - lastDownloaded); lastDownloaded = counter.downloaded; + + if (retryPolicy != null && (contentLength < 0 || counter.downloaded < contentLength) + && retryPolicy.shouldRetry(counter.downloaded, System.nanoTime())) + throw new SocketTimeoutException("BMCLAPI node remained below 128 KiB/s for 10 seconds"); } if (isCancelled()) @@ -308,6 +331,7 @@ private void downloadHttp(WebURL url, int retry, Duration connectTimeout, Durati @Nullable Context context = null; @Nullable HttpResumeContext resumeContext = null; + BMCLAPIRetryPolicy retryPolicy = new BMCLAPIRetryPolicy(); @Nullable ArrayList exceptions = null; @@ -326,7 +350,7 @@ private void downloadHttp(WebURL url, int retry, Duration connectTimeout, Durati @Nullable HttpURLConnection connection; UrlResponseInfo responseInfo; - @Nullable String bmclapiHash; + @Nullable String bmclapiHash = null; int responseCode; WebURL currentURI = url; @@ -339,7 +363,9 @@ private void downloadHttp(WebURL url, int retry, Duration connectTimeout, Durati headers.putAll(repository.injectConnection(url)); if (resumeRequested) { headers.put("range", "bytes=" + resumeContext.countUncompressed + "-"); - headers.put("if-range", resumeContext.ifRange()); + @Nullable String ifRange = resumeContext.ifRange(); + if (ifRange != null) + headers.put("if-range", ifRange); } do { @@ -352,8 +378,11 @@ private void downloadHttp(WebURL url, int retry, Duration connectTimeout, Durati responseCode = connection.getResponseCode(); responseInfo = UrlResponseInfo.of(connection); - bmclapiHash = responseInfo.headers().firstValue("x-bmclapi-hash").orElse(null); - if (DigestUtils.isSha1Digest(bmclapiHash)) { + @Nullable String responseHash = responseInfo.headers().firstValue("x-bmclapi-hash").orElse(null); + if (DigestUtils.isSha1Digest(responseHash)) { + if (bmclapiHash != null && !bmclapiHash.equalsIgnoreCase(responseHash)) + throw new IOException("Conflicting BMCLAPI hashes in redirect chain"); + bmclapiHash = responseHash; Optional cache = repository.checkExistentFile(null, "SHA-1", bmclapiHash); if (cache.isPresent()) { useCachedResult(cache.get()); @@ -432,9 +461,9 @@ private void downloadHttp(WebURL url, int retry, Duration connectTimeout, Durati if (context == null) { context = getContext(responseInfo, checkETag, bmclapiHash); - resumeContext = HttpResumeContext.of(responseInfo); + resumeContext = HttpResumeContext.of(responseInfo, bmclapiHash); } else if (resumeRequested) { - if (resumeContext.canResume(responseCode, responseInfo)) { + if (resumeContext.canResume(responseCode, responseInfo, bmclapiHash)) { // Resume download LOG.info("Resuming " + resumeContext.url + " from " + resumeContext.countUncompressed); } else { @@ -448,7 +477,7 @@ private void downloadHttp(WebURL url, int retry, Duration connectTimeout, Durati } else { discardContext(context); context = getContext(responseInfo, checkETag, bmclapiHash); - resumeContext = HttpResumeContext.of(responseInfo); + resumeContext = HttpResumeContext.of(responseInfo, bmclapiHash); } try { @@ -456,7 +485,8 @@ private void downloadHttp(WebURL url, int retry, Duration connectTimeout, Durati download(context, resumeContext, inputStream, contentLength, - contentEncoding); + contentEncoding, bmclapiHash != null && redirects != null && attempts < attemptsLimit - 1 + ? retryPolicy : null); inputStream = null; responseBodyConsumed = true; } catch (IOException | InterruptedException | RuntimeException | Error e) { @@ -547,7 +577,7 @@ private void downloadNotHttp(WebURL url, int retry, Duration connectTimeout, Dur download(context, null, conn.getInputStream(), conn.getContentLengthLong(), - ContentEncoding.fromConnection(conn)); + ContentEncoding.fromConnection(conn), null); } return; } catch (InterruptedException e) { @@ -602,6 +632,47 @@ private static void updateDownloadSpeed(long speed) { downloadSpeed.addAndGet(speed); } + /// Retries persistently slow BMCLAPI nodes without exhausting downloads on bandwidth-limited connections. + static final class BMCLAPIRetryPolicy { + /// Maximum slow-node retries within one candidate's existing retry budget. + private static final int MAX_SLOW_RETRIES = 2; + /// Observation window for each connection. + private static final long WINDOW_NANOS = TimeUnit.SECONDS.toNanos(10); + /// Minimum transfer rate in bytes per second before retrying a slow node. + private static final long MIN_BYTES_PER_SECOND = 128 * 1024; + /// Monotonic time at the beginning of the current window. + private long started; + /// Total bytes received before the current window. + private long downloaded; + /// Number of node reselections requested by this policy across all attempts. + private int retries; + + /// Starts a fresh observation window while retaining the candidate's slow-node retry count. + void startAttempt(long now) { + this.started = now; + this.downloaded = 0; + } + + /// Checks a complete sampling window and consumes one retry when its average rate is too low. + boolean shouldRetry(long downloaded, long now) { + if (retries >= MAX_SLOW_RETRIES) + return false; + + long elapsed = now - started; + if (elapsed < WINDOW_NANOS) + return false; + + double bytesPerSecond = (downloaded - this.downloaded) * (double) TimeUnit.SECONDS.toNanos(1) / elapsed; + this.started = now; + this.downloaded = downloaded; + if (bytesPerSecond >= MIN_BYTES_PER_SECOND) + return false; + + retries++; + return true; + } + } + private static final class CounterInputStream extends FilterInputStream { long downloaded; diff --git a/HMCLCore/src/main/java/org/jackhuang/hmcl/task/GetTask.java b/HMCLCore/src/main/java/org/jackhuang/hmcl/task/GetTask.java index 5bfd21899f2..e90c92dcf9d 100644 --- a/HMCLCore/src/main/java/org/jackhuang/hmcl/task/GetTask.java +++ b/HMCLCore/src/main/java/org/jackhuang/hmcl/task/GetTask.java @@ -20,7 +20,9 @@ import com.google.gson.reflect.TypeToken; import org.glavo.url.WebURL; import org.jackhuang.hmcl.download.DownloadCandidates; +import org.jackhuang.hmcl.util.DigestUtils; import org.jackhuang.hmcl.util.gson.JsonUtils; +import org.jackhuang.hmcl.util.io.ChecksumMismatchException; import org.jackhuang.hmcl.util.io.NetworkUtils; import org.jackhuang.hmcl.util.io.UrlResponseInfo; import org.jetbrains.annotations.Nullable; @@ -63,6 +65,7 @@ protected void useCachedResult(Path cachedFile) throws IOException { setResult(Files.readString(cachedFile)); } + /// Creates an in-memory sink, verifying the supplied content hash before decoding or caching text. @Override protected Context getContext(@Nullable UrlResponseInfo response, boolean checkETag, @Nullable String bmclapiHash) { long length = -1; @@ -85,6 +88,12 @@ public void write(byte[] buffer, int offset, int len) { public void close() throws IOException { if (!isSuccess()) return; + if (bmclapiHash != null) { + String actualHash = DigestUtils.digestToString("SHA-1", baos.toByteArray()); + if (!bmclapiHash.equalsIgnoreCase(actualHash)) + throw new ChecksumMismatchException("SHA-1", bmclapiHash, actualHash); + } + Charset charset = StandardCharsets.UTF_8; if (response != null) charset = NetworkUtils.getCharsetFromContentType(response.headers().firstValue("content-type").orElse(null)); diff --git a/HMCLCore/src/test/java/org/jackhuang/hmcl/task/BMCLAPIRetryPolicyTest.java b/HMCLCore/src/test/java/org/jackhuang/hmcl/task/BMCLAPIRetryPolicyTest.java new file mode 100644 index 00000000000..38091ad46d0 --- /dev/null +++ b/HMCLCore/src/test/java/org/jackhuang/hmcl/task/BMCLAPIRetryPolicyTest.java @@ -0,0 +1,81 @@ +/* + * Hello Minecraft! Launcher + * Copyright (C) 2026 huangyuhui and contributors + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ +package org.jackhuang.hmcl.task; + +import org.jetbrains.annotations.NotNullByDefault; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/// Tests node retry decisions using elapsed time values without waiting for the wall clock. +@NotNullByDefault +public final class BMCLAPIRetryPolicyTest { + /// One second expressed in the policy's monotonic time unit. + private static final long SECOND = TimeUnit.SECONDS.toNanos(1); + + /// A new connection receives a complete observation window before it can be retried. + @Test + public void waitsForCompleteObservationWindow() { + var policy = new FetchTask.BMCLAPIRetryPolicy(); + policy.startAttempt(SECOND); + assertFalse(policy.shouldRetry(1, 10 * SECOND)); + assertTrue(policy.shouldRetry(1, 11 * SECOND)); + } + + /// An earlier fast window does not hide a later persistently slow window. + @Test + public void measuresSuccessiveWindowsIndependently() { + var policy = new FetchTask.BMCLAPIRetryPolicy(); + policy.startAttempt(0); + long downloaded = 128 * 1024 * 10; + assertFalse(policy.shouldRetry(downloaded, 10 * SECOND)); + assertFalse(policy.shouldRetry(downloaded + 1, 19 * SECOND)); + assertTrue(policy.shouldRetry(downloaded + 1, 20 * SECOND)); + } + + /// Delayed samples use their actual elapsed duration instead of the nominal window length. + @Test + public void accountsForDelayedSamples() { + var policy = new FetchTask.BMCLAPIRetryPolicy(); + policy.startAttempt(0); + assertTrue(policy.shouldRetry(128 * 1024 * 10, 20 * SECOND)); + } + + /// Healthy windows preserve the retry budget, which remains bounded across new connections. + @Test + public void retainsRetryBudgetAcrossAttempts() { + var policy = new FetchTask.BMCLAPIRetryPolicy(); + policy.startAttempt(0); + long windowBytes = 128 * 1024 * 10; + for (int window = 1; window <= 5; window++) { + assertFalse(policy.shouldRetry(windowBytes * window, window * 10 * SECOND)); + } + assertTrue(policy.shouldRetry(windowBytes * 5 + 1, 60 * SECOND)); + + policy.startAttempt(100 * SECOND); + assertFalse(policy.shouldRetry(1, 109 * SECOND)); + assertTrue(policy.shouldRetry(1, 110 * SECOND)); + + policy.startAttempt(200 * SECOND); + assertFalse(policy.shouldRetry(1, 210 * SECOND)); + assertFalse(policy.shouldRetry(1, 220 * SECOND)); + } +} diff --git a/HMCLCore/src/test/java/org/jackhuang/hmcl/task/FetchTaskTest.java b/HMCLCore/src/test/java/org/jackhuang/hmcl/task/FetchTaskTest.java new file mode 100644 index 00000000000..10adee14612 --- /dev/null +++ b/HMCLCore/src/test/java/org/jackhuang/hmcl/task/FetchTaskTest.java @@ -0,0 +1,396 @@ +/* + * Hello Minecraft! Launcher + * Copyright (C) 2026 huangyuhui and contributors + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ +package org.jackhuang.hmcl.task; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpHandler; +import com.sun.net.httpserver.HttpServer; +import org.glavo.url.WebURL; +import org.jackhuang.hmcl.download.DownloadCandidate; +import org.jackhuang.hmcl.download.DownloadCandidates; +import org.jackhuang.hmcl.util.CacheRepository; +import org.jackhuang.hmcl.util.DigestUtils; +import org.jetbrains.annotations.NotNullByDefault; +import org.jetbrains.annotations.Nullable; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Objects; +import java.util.Random; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.*; + +/// Tests sequential downloads, node reselection, and resume integrity using local HTTP servers. +@NotNullByDefault +@Timeout(40) +public final class FetchTaskTest { + /// Directory containing this test's download target and cache. + @TempDir + Path directory; + + /// Reselects a slow node and resumes through an object-storage redirect without a final hash header. + @Test + public void reselectsSlowNodeAndResumesThroughStorageRedirect() throws Exception { + byte[] content = content(2 * 1024 * 1024); + String hash = sha1(content); + AtomicInteger dispatches = new AtomicInteger(); + AtomicInteger slowRequests = new AtomicInteger(); + CopyOnWriteArrayList ranges = new CopyOnWriteArrayList<>(); + CopyOnWriteArrayList ifRanges = new CopyOnWriteArrayList<>(); + try (TestServer server = new TestServer()) { + server.handle("/master", exchange -> { + int attempt = dispatches.incrementAndGet(); + redirect(exchange, attempt == 1 ? "/slow" : "/node", hash); + }); + server.handle("/slow", exchange -> { + slowRequests.incrementAndGet(); + exchange.getResponseHeaders().set("Accept-Ranges", "bytes"); + exchange.getResponseHeaders().set("ETag", "W/\"node-a\""); + exchange.sendResponseHeaders(200, content.length); + try { + for (int offset = 0; offset < content.length; offset += 4096) { + exchange.getResponseBody().write(content, offset, Math.min(4096, content.length - offset)); + exchange.getResponseBody().flush(); + TimeUnit.MILLISECONDS.sleep(100); + } + } catch (IOException ignored) { + // The client closes this connection after detecting a slow node. + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + server.handle("/node", exchange -> redirect(exchange, "/storage", hash)); + server.handle("/storage", exchange -> { + @Nullable String range = exchange.getRequestHeaders().getFirst("Range"); + ranges.add(Objects.requireNonNullElse(range, "")); + ifRanges.add(Objects.requireNonNullElse(exchange.getRequestHeaders().getFirst("If-Range"), "")); + int offset = rangeOffset(range); + exchange.getResponseHeaders().set("ETag", "\"node-b\""); + sendContent(exchange, content, offset, content.length - offset); + }); + + Path target = download(server.url("/master")); + assertArrayEquals(content, Files.readAllBytes(target)); + assertEquals(2, dispatches.get()); + assertEquals(1, slowRequests.get()); + assertEquals(1, ranges.size()); + assertTrue(ranges.get(0).startsWith("bytes=")); + assertTrue(rangeOffset(ranges.get(0)) > 0); + assertEquals("", ifRanges.get(0)); + } + } + + /// Validates hashes, ranges, and the completed digest, discarding partial content after invalid responses. + @ParameterizedTest + @EnumSource(ResumeResponse.class) + public void validatesResumptionAcrossNodes(ResumeResponse responseType) throws Exception { + byte[] content = content(64 * 1024); + String hash = sha1(content); + AtomicInteger dispatches = new AtomicInteger(); + AtomicInteger entries = new AtomicInteger(); + CopyOnWriteArrayList ranges = new CopyOnWriteArrayList<>(); + CopyOnWriteArrayList ifRanges = new CopyOnWriteArrayList<>(); + try (TestServer server = new TestServer()) { + server.handle("/entry", exchange -> { + entries.incrementAndGet(); + redirect(exchange, "/master", null); + }); + server.handle("/master", exchange -> { + int attempt = dispatches.incrementAndGet(); + String selectedHash = responseType == ResumeResponse.CHANGED_HASH && attempt == 2 ? "0".repeat(40) : hash; + redirect(exchange, "/node" + attempt, selectedHash); + }); + server.handle("/node", exchange -> { + int attempt = Integer.parseInt(exchange.getRequestURI().getPath().substring("/node".length())); + @Nullable String range = exchange.getRequestHeaders().getFirst("Range"); + ranges.add(Objects.requireNonNullElse(range, "")); + ifRanges.add(Objects.requireNonNullElse(exchange.getRequestHeaders().getFirst("If-Range"), "")); + if (attempt == 1) { + // This node provides a content hash and byte ranges without reusable HTTP validators. + sendContent(exchange, content, 0, 4096); + } else if (attempt == 2 && responseType == ResumeResponse.UNSATISFIABLE_RANGE) { + exchange.sendResponseHeaders(416, -1); + } else { + int offset = responseType == ResumeResponse.IGNORED_RANGE ? 0 : rangeOffset(range); + if (responseType == ResumeResponse.INVALID_RANGE && attempt == 2) + offset++; + byte[] response = content.clone(); + if (responseType == ResumeResponse.CORRUPT_BODY && attempt == 2) + response[response.length - 1] ^= 1; + sendContent(exchange, response, offset, response.length - offset); + } + }); + + Path target = download(server.url("/entry")); + assertArrayEquals(content, Files.readAllBytes(target)); + assertEquals(dispatches.get(), entries.get()); + assertEquals(responseType == ResumeResponse.VALID ? 2 : 3, dispatches.get()); + assertEquals("", ranges.get(0)); + assertEquals("bytes=4096-", ranges.get(1)); + assertTrue(ifRanges.stream().allMatch(String::isEmpty)); + if (responseType != ResumeResponse.VALID) + assertEquals("", ranges.get(2)); + } + } + + /// Rejects corrupt content even when the final response omits the redirect chain's hash. + @Test + public void retainsHashUntilFinalResponse() throws Exception { + byte[] content = content(8192); + String hash = sha1(content); + AtomicInteger requests = new AtomicInteger(); + try (TestServer server = new TestServer()) { + server.handle("/master", exchange -> redirect(exchange, "/storage", hash)); + server.handle("/storage", exchange -> { + byte[] response = content.clone(); + if (requests.incrementAndGet() == 1) + response[0] ^= 1; + sendContent(exchange, response, 0, response.length); + }); + Path target = download(server.url("/master")); + assertArrayEquals(content, Files.readAllBytes(target)); + assertEquals(2, requests.get()); + } + } + + /// Preserves strong ETag and If-Range validation for ordinary HTTP downloads. + @Test + public void preservesStandardHttpResumption() throws Exception { + byte[] content = content(64 * 1024); + AtomicInteger requests = new AtomicInteger(); + CopyOnWriteArrayList ifRanges = new CopyOnWriteArrayList<>(); + try (TestServer server = new TestServer()) { + server.handle("/file", exchange -> { + int attempt = requests.incrementAndGet(); + ifRanges.add(Objects.requireNonNullElse(exchange.getRequestHeaders().getFirst("If-Range"), "")); + exchange.getResponseHeaders().set("ETag", "\"stable\""); + int offset = rangeOffset(exchange.getRequestHeaders().getFirst("Range")); + sendContent(exchange, content, offset, attempt == 1 ? 4096 : content.length - offset); + }); + assertArrayEquals(content, Files.readAllBytes(download(server.url("/file")))); + assertEquals(2, requests.get()); + assertEquals("\"stable\"", ifRanges.get(1)); + } + } + + /// Verifies resumed cache content even when the cache task has no explicit expected checksum. + @Test + public void verifiesResumedCacheContentBeforePublishing() throws Exception { + byte[] content = content(64 * 1024); + String hash = sha1(content); + AtomicInteger requests = new AtomicInteger(); + try (TestServer server = new TestServer()) { + server.handle("/master", exchange -> redirect(exchange, "/node", hash)); + server.handle("/node", exchange -> { + int attempt = requests.incrementAndGet(); + byte[] response = content.clone(); + if (attempt == 2) + response[response.length - 1] ^= 1; + exchange.getResponseHeaders().set("ETag", "W/\"cache-node\""); + int offset = rangeOffset(exchange.getRequestHeaders().getFirst("Range")); + sendContent(exchange, response, offset, attempt == 1 ? 4096 : response.length - offset); + }); + CacheFileTask task = new CacheFileTask(server.url("/master")); + configureRepository(task); + task.execute(); + assertArrayEquals(content, Files.readAllBytes(task.getResult())); + assertEquals(3, requests.get()); + } + } + + /// Verifies text before decoding, even when the response cannot be resumed. + @Test + public void verifiesUnresumableTextContent() throws Exception { + String text = "BMCLAPI content integrity"; + byte[] content = text.getBytes(StandardCharsets.UTF_8); + String hash = sha1(content); + AtomicInteger requests = new AtomicInteger(); + try (TestServer server = new TestServer()) { + server.handle("/master", exchange -> redirect(exchange, "/node", hash)); + server.handle("/node", exchange -> { + byte[] response = content.clone(); + if (requests.incrementAndGet() == 1) + response[0] ^= 1; + exchange.sendResponseHeaders(200, response.length); + exchange.getResponseBody().write(response); + }); + GetTask task = new GetTask(server.url("/master")); + configureRepository(task); + task.execute(); + assertEquals(text, task.getResult()); + assertEquals(2, requests.get()); + } + } + + /// Verifies cache content independently of the server's range support. + @Test + public void verifiesUnresumableCacheContent() throws Exception { + byte[] content = content(8192); + String hash = sha1(content); + AtomicInteger requests = new AtomicInteger(); + try (TestServer server = new TestServer()) { + server.handle("/master", exchange -> redirect(exchange, "/node", hash)); + server.handle("/node", exchange -> { + byte[] response = content.clone(); + if (requests.incrementAndGet() == 1) + response[0] ^= 1; + exchange.getResponseHeaders().set("ETag", "\"cache-node\""); + exchange.sendResponseHeaders(200, response.length); + exchange.getResponseBody().write(response); + }); + CacheFileTask task = new CacheFileTask(server.url("/master")); + configureRepository(task); + task.execute(); + assertArrayEquals(content, Files.readAllBytes(task.getResult())); + assertEquals(2, requests.get()); + } + } + + /// Downloads a file using an isolated cache and a bounded retry budget. + private Path download(WebURL url) throws Exception { + Path target = directory.resolve("download.bin"); + DownloadCandidates candidates = DownloadCandidates.of(new DownloadCandidate(url, null, 3, + Duration.ofSeconds(2), Duration.ofSeconds(2))); + FileDownloadTask task = new FileDownloadTask(candidates, target); + configureRepository(task); + task.execute(); + return target; + } + + /// Keeps the task's cache within the current test directory. + private void configureRepository(FetchTask task) { + CacheRepository repository = new CacheRepository(); + repository.changeDirectory(directory); + task.setCacheRepository(repository); + } + + /// Creates reproducible, incompressible file content. + private static byte[] content(int size) { + byte[] result = new byte[size]; + new Random(7002).nextBytes(result); + return result; + } + + /// Computes the content hash supplied by the simulated dispatcher. + private static String sha1(byte[] content) { + return DigestUtils.digestToString("SHA-1", content); + } + + /// Sends a dispatcher or node redirect with an optional content hash. + private static void redirect(HttpExchange exchange, String location, @Nullable String hash) throws IOException { + exchange.getResponseHeaders().set("Location", location); + if (hash != null) + exchange.getResponseHeaders().set("X-BMCLAPI-Hash", hash); + exchange.sendResponseHeaders(302, -1); + } + + /// Parses a single open-ended byte range used by these tests. + private static int rangeOffset(@Nullable String range) { + return range == null ? 0 : Integer.parseInt(range.substring("bytes=".length(), range.length() - 1)); + } + + /// Sends full or partial content, optionally truncating the body to simulate an interrupted download. + private static void sendContent(HttpExchange exchange, byte[] content, int offset, int count) throws IOException { + exchange.getResponseHeaders().set("Accept-Ranges", "bytes"); + int length = content.length - offset; + if (offset > 0) + exchange.getResponseHeaders().set("Content-Range", "bytes " + offset + "-" + (content.length - 1) + "/" + content.length); + exchange.sendResponseHeaders(offset > 0 ? 206 : 200, length); + exchange.getResponseBody().write(content, offset, count); + } + + /// Response behaviors relevant to cross-node resumption. + @NotNullByDefault + private enum ResumeResponse { + /// Matching content hash and byte range. + VALID, + /// A complete response instead of the requested range. + IGNORED_RANGE, + /// A byte range starting at the wrong offset. + INVALID_RANGE, + /// A different content hash advertised by the dispatcher. + CHANGED_HASH, + /// A matching range whose body fails integrity checking. + CORRUPT_BODY, + /// A server rejecting the requested range with HTTP 416. + UNSATISFIABLE_RANGE + } + + /// Handles requests independently, allowing tests to observe requests made before an old response finishes. + @NotNullByDefault + private static final class TestServer implements AutoCloseable { + /// Local HTTP server listening only on the loopback interface. + private final HttpServer server; + /// First unexpected handler failure, propagated to the test thread on close. + private final AtomicReference<@Nullable Throwable> failure = new AtomicReference<>(); + /// Executor for responses, stopped when the test server is closed. + private final ExecutorService executor = Executors.newCachedThreadPool(); + + /// Starts a server on an automatically assigned port. + TestServer() throws IOException { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.setExecutor(executor); + server.start(); + } + + /// Registers a handler that closes its exchange after responding. + public void handle(String path, HttpHandler handler) { + server.createContext(path, exchange -> { + try { + handler.handle(exchange); + } catch (Throwable e) { + failure.compareAndSet(null, e); + throw e; + } finally { + exchange.close(); + } + }); + } + + /// Creates a download URL for this server. + WebURL url(String path) { + return WebURL.parse("http://127.0.0.1:" + server.getAddress().getPort() + path); + } + + /// Releases the listening port, stops handlers, and reports their unexpected failures. + @Override + public void close() throws InterruptedException { + server.stop(0); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS), "HTTP handlers did not stop"); + @Nullable Throwable error = failure.get(); + if (error != null) + throw new AssertionError("HTTP handler failed", error); + } + } +} From 910cb02d4641f006579aa895aeebbbe5985f96e6 Mon Sep 17 00:00:00 2001 From: wyjh Date: Tue, 20 Oct 2026 23:48:35 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E8=B7=A8=E8=8A=82?= =?UTF-8?q?=E7=82=B9=E7=BB=AD=E4=BC=A0=E5=B9=B6=E6=8C=89=E5=89=A9=E4=BD=99?= =?UTF-8?q?=E4=B8=8B=E8=BD=BD=E6=97=B6=E9=97=B4=E5=88=A4=E6=96=AD=E4=BD=8E?= =?UTF-8?q?=E9=80=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../org/jackhuang/hmcl/task/FetchTask.java | 38 ++++++------ .../hmcl/task/BMCLAPIRetryPolicyTest.java | 57 ++++++++++++++---- .../jackhuang/hmcl/task/FetchTaskTest.java | 58 +++++++++++++++++-- 3 files changed, 118 insertions(+), 35 deletions(-) diff --git a/HMCLCore/src/main/java/org/jackhuang/hmcl/task/FetchTask.java b/HMCLCore/src/main/java/org/jackhuang/hmcl/task/FetchTask.java index 1215450a9d8..9eb22e4a52a 100644 --- a/HMCLCore/src/main/java/org/jackhuang/hmcl/task/FetchTask.java +++ b/HMCLCore/src/main/java/org/jackhuang/hmcl/task/FetchTask.java @@ -148,15 +148,11 @@ public void execute() throws Exception { private static final class HttpResumeContext { private static final Pattern CONTENT_RANGE_PATTERN = Pattern.compile("bytes ([0-9]+)-([0-9]+)/([0-9]+)"); - /// Creates resume state for an uncompressed response with a known length and byte-range support. + /// Creates resume state for an uncompressed response with a known length and a file validator. static @Nullable FetchTask.HttpResumeContext of(UrlResponseInfo response, @Nullable String bmclapiHash) throws IOException { if (response.responseCode() != HttpURLConnection.HTTP_OK) return null; - boolean acceptRanges = response.headers().firstValue("accept-ranges").orElse("").equalsIgnoreCase("bytes"); - if (!acceptRanges) - return null; - var contentEncoding = ContentEncoding.fromHeaders(response.headers()); if (contentEncoding != ContentEncoding.IDENTITY) return null; @@ -216,7 +212,7 @@ boolean canResume(int statusCode, UrlResponseInfo response, @Nullable String bmc return false; long contentLength = response.headers().firstValueAsLong("content-length").orElse(-1L); - if (this.contentLength != contentLength + this.countUncompressed) + if (contentLength >= 0 && this.contentLength - this.countUncompressed != contentLength) return false; if (this.bmclapiHash != null) { @@ -245,10 +241,7 @@ boolean canResume(int statusCode, UrlResponseInfo response, @Nullable String bmc long end = Long.parseLong(matcher.group(2)); long total = Long.parseLong(matcher.group(3)); - if (start != countUncompressed || end < start || total != this.contentLength) - return false; - - return end - start + 1 == contentLength; + return start == countUncompressed && total == this.contentLength && end == total - 1; } catch (NumberFormatException e) { return false; } @@ -287,7 +280,9 @@ private void download(Context context, if (resume != null) resume.countUncompressed += len; - if (contentLength >= 0) { + if (resume != null) { + updateProgress(resume.countUncompressed, resume.contentLength); + } else if (contentLength >= 0) { // Update progress information per second updateProgress(counter.downloaded, contentLength); } @@ -295,9 +290,8 @@ private void download(Context context, updateDownloadSpeed(counter.downloaded - lastDownloaded); lastDownloaded = counter.downloaded; - if (retryPolicy != null && (contentLength < 0 || counter.downloaded < contentLength) - && retryPolicy.shouldRetry(counter.downloaded, System.nanoTime())) - throw new SocketTimeoutException("BMCLAPI node remained below 128 KiB/s for 10 seconds"); + if (retryPolicy != null && retryPolicy.shouldRetry(counter.downloaded, contentLength, System.nanoTime())) + throw new SocketTimeoutException("BMCLAPI node is too slow to complete the remaining download promptly"); } if (isCancelled()) @@ -346,7 +340,8 @@ private void downloadHttp(WebURL url, int retry, Duration connectTimeout, Durati @Nullable List redirects = null; try { beforeDownload(url); - updateProgress(0); + if (resumeContext == null || !resumeContext.hasPartialContent()) + updateProgress(0); @Nullable HttpURLConnection connection; UrlResponseInfo responseInfo; @@ -362,6 +357,7 @@ private void downloadHttp(WebURL url, int retry, Duration connectTimeout, Durati if (useCachedResult && checkETag && !resumeRequested) headers.putAll(repository.injectConnection(url)); if (resumeRequested) { + headers.put("accept-encoding", "identity"); headers.put("range", "bytes=" + resumeContext.countUncompressed + "-"); @Nullable String ifRange = resumeContext.ifRange(); if (ifRange != null) @@ -383,6 +379,7 @@ private void downloadHttp(WebURL url, int retry, Duration connectTimeout, Durati if (bmclapiHash != null && !bmclapiHash.equalsIgnoreCase(responseHash)) throw new IOException("Conflicting BMCLAPI hashes in redirect chain"); bmclapiHash = responseHash; + headers.put("accept-encoding", "identity"); Optional cache = repository.checkExistentFile(null, "SHA-1", bmclapiHash); if (cache.isPresent()) { useCachedResult(cache.get()); @@ -466,6 +463,7 @@ private void downloadHttp(WebURL url, int retry, Duration connectTimeout, Durati if (resumeContext.canResume(responseCode, responseInfo, bmclapiHash)) { // Resume download LOG.info("Resuming " + resumeContext.url + " from " + resumeContext.countUncompressed); + contentLength = resumeContext.contentLength - resumeContext.countUncompressed; } else { // Failed to resume download, so we will retry from the beginning resumeContext = null; @@ -653,9 +651,10 @@ void startAttempt(long now) { this.downloaded = 0; } - /// Checks a complete sampling window and consumes one retry when its average rate is too low. - boolean shouldRetry(long downloaded, long now) { - if (retries >= MAX_SLOW_RETRIES) + /// Retries a slow window only when finishing at the current rate would take another full window. + /// Unknown-length responses cannot provide a useful estimate and retain normal timeout handling. + boolean shouldRetry(long downloaded, long contentLength, long now) { + if (retries >= MAX_SLOW_RETRIES || contentLength <= downloaded) return false; long elapsed = now - started; @@ -665,7 +664,8 @@ boolean shouldRetry(long downloaded, long now) { double bytesPerSecond = (downloaded - this.downloaded) * (double) TimeUnit.SECONDS.toNanos(1) / elapsed; this.started = now; this.downloaded = downloaded; - if (bytesPerSecond >= MIN_BYTES_PER_SECOND) + if (bytesPerSecond >= MIN_BYTES_PER_SECOND + || contentLength - downloaded <= bytesPerSecond * WINDOW_NANOS / TimeUnit.SECONDS.toNanos(1)) return false; retries++; diff --git a/HMCLCore/src/test/java/org/jackhuang/hmcl/task/BMCLAPIRetryPolicyTest.java b/HMCLCore/src/test/java/org/jackhuang/hmcl/task/BMCLAPIRetryPolicyTest.java index 38091ad46d0..f1cdd44b085 100644 --- a/HMCLCore/src/test/java/org/jackhuang/hmcl/task/BMCLAPIRetryPolicyTest.java +++ b/HMCLCore/src/test/java/org/jackhuang/hmcl/task/BMCLAPIRetryPolicyTest.java @@ -30,14 +30,16 @@ public final class BMCLAPIRetryPolicyTest { /// One second expressed in the policy's monotonic time unit. private static final long SECOND = TimeUnit.SECONDS.toNanos(1); + /// A large remaining download used when only the sampling window is under test. + private static final long CONTENT_LENGTH = 100 * 1024 * 1024; /// A new connection receives a complete observation window before it can be retried. @Test public void waitsForCompleteObservationWindow() { var policy = new FetchTask.BMCLAPIRetryPolicy(); policy.startAttempt(SECOND); - assertFalse(policy.shouldRetry(1, 10 * SECOND)); - assertTrue(policy.shouldRetry(1, 11 * SECOND)); + assertFalse(policy.shouldRetry(1, CONTENT_LENGTH, 10 * SECOND)); + assertTrue(policy.shouldRetry(1, CONTENT_LENGTH, 11 * SECOND)); } /// An earlier fast window does not hide a later persistently slow window. @@ -46,9 +48,9 @@ public void measuresSuccessiveWindowsIndependently() { var policy = new FetchTask.BMCLAPIRetryPolicy(); policy.startAttempt(0); long downloaded = 128 * 1024 * 10; - assertFalse(policy.shouldRetry(downloaded, 10 * SECOND)); - assertFalse(policy.shouldRetry(downloaded + 1, 19 * SECOND)); - assertTrue(policy.shouldRetry(downloaded + 1, 20 * SECOND)); + assertFalse(policy.shouldRetry(downloaded, CONTENT_LENGTH, 10 * SECOND)); + assertFalse(policy.shouldRetry(downloaded + 1, CONTENT_LENGTH, 19 * SECOND)); + assertTrue(policy.shouldRetry(downloaded + 1, CONTENT_LENGTH, 20 * SECOND)); } /// Delayed samples use their actual elapsed duration instead of the nominal window length. @@ -56,7 +58,7 @@ public void measuresSuccessiveWindowsIndependently() { public void accountsForDelayedSamples() { var policy = new FetchTask.BMCLAPIRetryPolicy(); policy.startAttempt(0); - assertTrue(policy.shouldRetry(128 * 1024 * 10, 20 * SECOND)); + assertTrue(policy.shouldRetry(128 * 1024 * 10, CONTENT_LENGTH, 20 * SECOND)); } /// Healthy windows preserve the retry budget, which remains bounded across new connections. @@ -66,16 +68,47 @@ public void retainsRetryBudgetAcrossAttempts() { policy.startAttempt(0); long windowBytes = 128 * 1024 * 10; for (int window = 1; window <= 5; window++) { - assertFalse(policy.shouldRetry(windowBytes * window, window * 10 * SECOND)); + assertFalse(policy.shouldRetry(windowBytes * window, CONTENT_LENGTH, window * 10 * SECOND)); } - assertTrue(policy.shouldRetry(windowBytes * 5 + 1, 60 * SECOND)); + assertTrue(policy.shouldRetry(windowBytes * 5 + 1, CONTENT_LENGTH, 60 * SECOND)); policy.startAttempt(100 * SECOND); - assertFalse(policy.shouldRetry(1, 109 * SECOND)); - assertTrue(policy.shouldRetry(1, 110 * SECOND)); + assertFalse(policy.shouldRetry(1, CONTENT_LENGTH, 109 * SECOND)); + assertTrue(policy.shouldRetry(1, CONTENT_LENGTH, 110 * SECOND)); policy.startAttempt(200 * SECOND); - assertFalse(policy.shouldRetry(1, 210 * SECOND)); - assertFalse(policy.shouldRetry(1, 220 * SECOND)); + assertFalse(policy.shouldRetry(1, CONTENT_LENGTH, 210 * SECOND)); + assertFalse(policy.shouldRetry(1, CONTENT_LENGTH, 220 * SECOND)); + } + + /// Small files and nearly completed large files finish without spending a node retry. + @Test + public void finishesRemainingContentBeforeReselecting() { + var policy = new FetchTask.BMCLAPIRetryPolicy(); + policy.startAttempt(0); + assertFalse(policy.shouldRetry(8 * 1024, 12 * 1024, 10 * SECOND)); + + policy.startAttempt(0); + assertFalse(policy.shouldRetry(64 * 1024 * 10, 64 * 1024 * 20, 10 * SECOND)); + assertFalse(policy.shouldRetry(64 * 1024 * 20, 64 * 1024 * 20, 20 * SECOND)); + + policy.startAttempt(0); + assertTrue(policy.shouldRetry(64 * 1024 * 10, CONTENT_LENGTH, 10 * SECOND)); + } + + /// Small files still reselect a node when their remaining transfer would take too long. + @Test + public void reselectsStalledSmallFiles() { + var policy = new FetchTask.BMCLAPIRetryPolicy(); + policy.startAttempt(0); + assertTrue(policy.shouldRetry(1024, 12 * 1024, 10 * SECOND)); + } + + /// A missing length cannot justify abandoning a response on an estimated completion time. + @Test + public void keepsUnknownLengthResponses() { + var policy = new FetchTask.BMCLAPIRetryPolicy(); + policy.startAttempt(0); + assertFalse(policy.shouldRetry(1, -1, 10 * SECOND)); } } diff --git a/HMCLCore/src/test/java/org/jackhuang/hmcl/task/FetchTaskTest.java b/HMCLCore/src/test/java/org/jackhuang/hmcl/task/FetchTaskTest.java index 10adee14612..13aa2410470 100644 --- a/HMCLCore/src/test/java/org/jackhuang/hmcl/task/FetchTaskTest.java +++ b/HMCLCore/src/test/java/org/jackhuang/hmcl/task/FetchTaskTest.java @@ -39,6 +39,8 @@ import java.nio.file.Path; import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.util.ArrayList; +import java.util.List; import java.util.Objects; import java.util.Random; import java.util.concurrent.CopyOnWriteArrayList; @@ -74,7 +76,6 @@ public void reselectsSlowNodeAndResumesThroughStorageRedirect() throws Exception }); server.handle("/slow", exchange -> { slowRequests.incrementAndGet(); - exchange.getResponseHeaders().set("Accept-Ranges", "bytes"); exchange.getResponseHeaders().set("ETag", "W/\"node-a\""); exchange.sendResponseHeaders(200, content.length); try { @@ -137,9 +138,18 @@ public void validatesResumptionAcrossNodes(ResumeResponse responseType) throws E ifRanges.add(Objects.requireNonNullElse(exchange.getRequestHeaders().getFirst("If-Range"), "")); if (attempt == 1) { // This node provides a content hash and byte ranges without reusable HTTP validators. - sendContent(exchange, content, 0, 4096); + if (responseType == ResumeResponse.OMITTED_ACCEPT_RANGES) { + exchange.sendResponseHeaders(200, content.length); + exchange.getResponseBody().write(content, 0, 4096); + } else { + sendContent(exchange, content, 0, 4096); + } } else if (attempt == 2 && responseType == ResumeResponse.UNSATISFIABLE_RANGE) { exchange.sendResponseHeaders(416, -1); + } else if (attempt == 2 && responseType == ResumeResponse.CHUNKED_RANGE) { + exchange.getResponseHeaders().set("Content-Range", "bytes 4096-" + (content.length - 1) + "/" + content.length); + exchange.sendResponseHeaders(206, 0); + exchange.getResponseBody().write(content, 4096, content.length - 4096); } else { int offset = responseType == ResumeResponse.IGNORED_RANGE ? 0 : rangeOffset(range); if (responseType == ResumeResponse.INVALID_RANGE && attempt == 2) @@ -154,11 +164,15 @@ public void validatesResumptionAcrossNodes(ResumeResponse responseType) throws E Path target = download(server.url("/entry")); assertArrayEquals(content, Files.readAllBytes(target)); assertEquals(dispatches.get(), entries.get()); - assertEquals(responseType == ResumeResponse.VALID ? 2 : 3, dispatches.get()); + boolean valid = switch (responseType) { + case VALID, OMITTED_ACCEPT_RANGES, CHUNKED_RANGE -> true; + default -> false; + }; + assertEquals(valid ? 2 : 3, dispatches.get()); assertEquals("", ranges.get(0)); assertEquals("bytes=4096-", ranges.get(1)); assertTrue(ifRanges.stream().allMatch(String::isEmpty)); - if (responseType != ResumeResponse.VALID) + if (!valid) assertEquals("", ranges.get(2)); } } @@ -203,6 +217,38 @@ public void preservesStandardHttpResumption() throws Exception { } } + /// Reports total file progress without resetting it when a partial response is appended. + @Test + public void preservesProgressAcrossResumption() throws Exception { + byte[] content = content(64 * 1024); + AtomicInteger requests = new AtomicInteger(); + List progressValues = new ArrayList<>(); + try (TestServer server = new TestServer()) { + server.handle("/file", exchange -> { + int attempt = requests.incrementAndGet(); + exchange.getResponseHeaders().set("ETag", "\"stable\""); + int offset = rangeOffset(exchange.getRequestHeaders().getFirst("Range")); + sendContent(exchange, content, offset, attempt == 1 ? 4096 : content.length - offset); + }); + Path target = directory.resolve("download.bin"); + FileDownloadTask task = new FileDownloadTask(List.of(server.url("/file")), target) { + /// Captures progress synchronously instead of dispatching JavaFX property updates. + @Override + protected void updateProgress(double progress) { + progressValues.add(progress); + } + }; + configureRepository(task); + task.execute(); + assertArrayEquals(content, Files.readAllBytes(target)); + assertEquals(2, requests.get()); + assertEquals(0.0, progressValues.get(0)); + assertEquals(1.0, progressValues.get(progressValues.size() - 1)); + for (int i = 1; i < progressValues.size(); i++) + assertTrue(progressValues.get(i) >= progressValues.get(i - 1), "Progress decreased: " + progressValues); + } + } + /// Verifies resumed cache content even when the cache task has no explicit expected checksum. @Test public void verifiesResumedCacheContentBeforePublishing() throws Exception { @@ -334,6 +380,10 @@ private static void sendContent(HttpExchange exchange, byte[] content, int offse private enum ResumeResponse { /// Matching content hash and byte range. VALID, + /// Range support was not advertised in the initial response. + OMITTED_ACCEPT_RANGES, + /// A valid partial response with chunked framing instead of Content-Length. + CHUNKED_RANGE, /// A complete response instead of the requested range. IGNORED_RANGE, /// A byte range starting at the wrong offset.