Skip to content

K2GO-376 fix(content-exchange): windowed transfer rate, not cumulative average #717

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
luisguzman-adfa merged 2 commits into from
Nov 1, 2026
Merged

K2GO-376 fix(content-exchange): windowed transfer rate, not cumulative average #717

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
dc64ebf
K2GO-376 fix(content-exchange): windowed transfer rate, not cumulativ…
luisguzman-adfa Oct 30, 2026
68f7b1a
K2GO-376 refactor(content-exchange): reuse DownloadEta for ETA; drop …
luisguzman-adfa Oct 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 19 additions & 13 deletions ...rc/main/java/org/appdevforall/k2go/contentexchange/presentation/ContentExportService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -238,8 +242,7 @@ private void drain(boolean fresh) {

ExportJob job = SESSION.job();
Set<String> 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()) {
Expand All @@ -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);
Expand All @@ -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();
Expand Down Expand Up @@ -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) {
Expand All @@ -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);
Expand Down Expand Up @@ -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);
}

Expand Down
32 changes: 19 additions & 13 deletions ...rc/main/java/org/appdevforall/k2go/contentexchange/presentation/ContentImportService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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()) {
Expand All @@ -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) {
Expand All @@ -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();
Expand Down Expand Up @@ -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;
Expand All @@ -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);
Expand All @@ -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);
}

Expand Down
Loading