diff --git a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ContentExportService.java b/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ContentExportService.java index 5d2dbc1eb..950bd1923 100644 --- a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ContentExportService.java +++ b/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ContentExportService.java @@ -42,6 +42,7 @@ import org.appdevforall.k2go.contentexchange.domain.ExportPrep; import org.appdevforall.k2go.contentexchange.domain.ExportStatus; import org.appdevforall.k2go.contentexchange.domain.ItemRef; +import org.appdevforall.k2go.download.domain.DownloadEta; import org.appdevforall.k2go.notifications.NotificationIds; import java.io.IOException; @@ -71,6 +72,9 @@ public final class ContentExportService extends Service { private static final int COPY_BUFFER = 64 * 1024; private static final long PUBLISH_INTERVAL_MS = 250L; private static final long NOTIFY_INTERVAL_MS = 1000L; + /** Minimum width of the trailing window used to sample the transfer rate (delta bytes / delta + * time), so the speed shown is the recent rate, not the cumulative average from the run start. */ + private static final long RATE_WINDOW_MS = 500L; /** K2GO-507: gap between export-staging status polls (mirrors content/RestContentClient). */ private static final long POLL_INTERVAL_MS = 1000L; /** K2GO-507: consecutive status-poll failures tolerated before an item fails. A staging run lasts @@ -238,8 +242,7 @@ private void drain(boolean fresh) { ExportJob job = SESSION.job(); Set used = seedUsedNames(); - long runStart = SystemClock.elapsedRealtime(); - long lastSampleAt = runStart; + long lastSampleAt = SystemClock.elapsedRealtime(); long lastSampleBytes = job.copiedTotal(); for (int i = job.firstPending(); i >= 0; i = job.firstPending()) { @@ -266,7 +269,7 @@ private void drain(boolean fresh) { // ("app/name"), which becomes the manifest payload. ExportCarrier.PayloadSink sink = carrier.openPayload(ref.app(), name); SESSION.setPayloadName(i, sink.name()); - copy(payload.stream(), sink.stream(), i, job, runStart); + copy(payload.stream(), sink.stream(), i, job); } // The core clears the op when the fetch stream closes, so the happy path does not abort. job.setDone(i); @@ -288,7 +291,7 @@ private void drain(boolean fresh) { } long now = SystemClock.elapsedRealtime(); - if (now - lastSampleAt >= 500L) { + if (now - lastSampleAt >= RATE_WINDOW_MS) { sampleRate(job, now - lastSampleAt, job.copiedTotal() - lastSampleBytes); lastSampleAt = now; lastSampleBytes = job.copiedTotal(); @@ -321,11 +324,16 @@ private void drain(boolean fresh) { * commonly surfaces at flush/close, not write) propagates and the item is marked failed, never * a silent partial written as done. */ - private void copy(InputStream in, OutputStream out, int i, ExportJob job, long runStart) + private void copy(InputStream in, OutputStream out, int i, ExportJob job) throws IOException { byte[] buffer = new byte[COPY_BUFFER]; long total = 0L; long lastPublish = 0L; + // Trailing-window rate: anchor at this item's copy start and re-anchor once the window is + // wide enough. A cumulative average (all bytes / whole run elapsed) reads far too low on a + // single large item that warms up slowly, so sample the recent throughput instead. + long windowAt = SystemClock.elapsedRealtime(); + long windowBytes = job.copiedTotal(); try (OutputStream sink = out) { int read; while ((read = in.read(buffer)) != -1) { @@ -337,12 +345,11 @@ private void copy(InputStream in, OutputStream out, int i, ExportJob job, long r job.setCopied(i, total); long now = SystemClock.elapsedRealtime(); if (now - lastPublish >= PUBLISH_INTERVAL_MS) { - long elapsed = now - runStart; - if (elapsed > 0) { - long speed = job.copiedTotal() * 1000L / elapsed; - long remaining = job.plannedTotal() - job.copiedTotal(); - long eta = speed > 0 ? Math.max(0L, remaining / speed) : -1L; - SESSION.setRate(speed, eta); + long windowMs = now - windowAt; + if (windowMs >= RATE_WINDOW_MS) { + sampleRate(job, windowMs, job.copiedTotal() - windowBytes); + windowAt = now; + windowBytes = job.copiedTotal(); } SESSION.publish(); maybeNotify(now); @@ -504,8 +511,7 @@ private void sampleRate(ExportJob job, long dtMs, long dBytes) { return; } long speed = Math.max(0L, dBytes * 1000L / dtMs); - long remaining = job.plannedTotal() - job.copiedTotal(); - long eta = speed > 0 ? Math.max(0L, remaining / speed) : -1L; + long eta = DownloadEta.secondsRemaining(job.copiedTotal(), job.plannedTotal(), speed); SESSION.setRate(speed, eta); } diff --git a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ContentImportService.java b/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ContentImportService.java index df3325bb4..fb1d7bb97 100644 --- a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ContentImportService.java +++ b/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ContentImportService.java @@ -40,6 +40,7 @@ import org.appdevforall.k2go.contentexchange.domain.ImportCarrier; import org.appdevforall.k2go.contentexchange.domain.ImportSource; import org.appdevforall.k2go.contentexchange.domain.ItemRef; +import org.appdevforall.k2go.download.domain.DownloadEta; import org.appdevforall.k2go.notifications.NotificationIds; import java.io.IOException; @@ -74,6 +75,9 @@ public final class ContentImportService extends Service { private static final int COPY_BUFFER = 64 * 1024; private static final long PUBLISH_INTERVAL_MS = 250L; private static final long NOTIFY_INTERVAL_MS = 1000L; + /** Minimum width of the trailing window used to sample the transfer rate (delta bytes / delta + * time), so the speed shown is the recent rate, not the cumulative average from the run start. */ + private static final long RATE_WINDOW_MS = 500L; /** The single run state the progress fragment observes. */ private static final ImportSession SESSION = new ImportSession(); @@ -283,8 +287,7 @@ private void drain() { SESSION.setSource(carrier.sourceLabel()); ExportJob job = SESSION.job(); - long runStart = SystemClock.elapsedRealtime(); - long lastSampleAt = runStart; + long lastSampleAt = SystemClock.elapsedRealtime(); long lastSampleBytes = job.copiedTotal(); for (int i = job.firstPending(); i >= 0; i = job.firstPending()) { @@ -298,7 +301,7 @@ private void drain() { String stored = BoxImportStaging.safeName(SESSION.payloadName(i)); try { - copyToStaging(carrier, staging, SESSION.payloadName(i), stored, i, job, runStart); + copyToStaging(carrier, staging, SESSION.payloadName(i), stored, i, job); source.importItem(ref.app(), SESSION.entryJson(i), BoxImportStaging.boxPath(stored)); job.setDone(i); } catch (Exception e) { @@ -316,7 +319,7 @@ private void drain() { staging.remove(stored); long now = SystemClock.elapsedRealtime(); - if (now - lastSampleAt >= 500L) { + if (now - lastSampleAt >= RATE_WINDOW_MS) { sampleRate(job, now - lastSampleAt, job.copiedTotal() - lastSampleBytes); lastSampleAt = now; lastSampleBytes = job.copiedTotal(); @@ -356,11 +359,16 @@ private void drain() { * failed, never a silent partial written as done. */ private void copyToStaging(ImportCarrier carrier, BoxImportStaging staging, String payloadName, - String storedName, int i, ExportJob job, long runStart) + String storedName, int i, ExportJob job) throws IOException { byte[] buffer = new byte[COPY_BUFFER]; long total = 0L; long lastPublish = 0L; + // Trailing-window rate: anchor at this item's copy start and re-anchor once the window is + // wide enough. A cumulative average (all bytes / whole run elapsed) reads far too low on a + // single large item that warms up slowly, so sample the recent throughput instead. + long windowAt = SystemClock.elapsedRealtime(); + long windowBytes = job.copiedTotal(); try (InputStream in = carrier.openPayload(payloadName); OutputStream out = staging.open(storedName)) { int read; @@ -373,12 +381,11 @@ private void copyToStaging(ImportCarrier carrier, BoxImportStaging staging, Stri job.setCopied(i, total); long now = SystemClock.elapsedRealtime(); if (now - lastPublish >= PUBLISH_INTERVAL_MS) { - long elapsed = now - runStart; - if (elapsed > 0) { - long speed = job.copiedTotal() * 1000L / elapsed; - long remaining = job.plannedTotal() - job.copiedTotal(); - long eta = speed > 0 ? Math.max(0L, remaining / speed) : -1L; - SESSION.setRate(speed, eta); + long windowMs = now - windowAt; + if (windowMs >= RATE_WINDOW_MS) { + sampleRate(job, windowMs, job.copiedTotal() - windowBytes); + windowAt = now; + windowBytes = job.copiedTotal(); } SESSION.publish(); maybeNotify(now); @@ -396,8 +403,7 @@ private void sampleRate(ExportJob job, long dtMs, long dBytes) { return; } long speed = Math.max(0L, dBytes * 1000L / dtMs); - long remaining = job.plannedTotal() - job.copiedTotal(); - long eta = speed > 0 ? Math.max(0L, remaining / speed) : -1L; + long eta = DownloadEta.secondsRemaining(job.copiedTotal(), job.plannedTotal(), speed); SESSION.setRate(speed, eta); }