From 6916c502181f4cfaa249c202f6a2ee490390cbc5 Mon Sep 17 00:00:00 2001 From: Kharkunov Eugene Date: Thu, 23 Jul 2026 10:43:52 +0300 Subject: [PATCH 1/2] Build progress tracking initial implementation --- README.md | 1 + README_BUILD_PROGRESS.md | 106 ++++++ README_CLIENT.md | 3 + .../extender/client/ExtenderClient.java | 32 +- .../client/ExtenderProgressConsumer.java | 167 ++++++++ .../client/ExtenderProgressListener.java | 24 ++ .../extender/client/ExtenderClientTest.java | 141 +++++++ .../com/defold/extender/AsyncBuilder.java | 18 + .../extender/BuildProgressController.java | 88 +++++ .../java/com/defold/extender/Extender.java | 27 +- .../defold/extender/ExtenderController.java | 18 + .../extender/process/ProcessExecutor.java | 8 + .../extender/progress/BuildProgressEvent.java | 47 +++ .../progress/BuildProgressService.java | 359 ++++++++++++++++++ .../defold/extender/progress/BuildStage.java | 25 ++ .../extender/progress/ProgressReporter.java | 30 ++ .../extender/remote/RemoteEngineBuilder.java | 35 +- .../extender/remote/RemoteProgressRelay.java | 188 +++++++++ server/src/main/resources/application.yml | 13 + .../extender/BuildProgressControllerTest.java | 209 ++++++++++ .../com/defold/extender/IntegrationTest.java | 94 +++++ .../extender/process/ProcessExecutorTest.java | 43 +++ .../progress/BuildProgressServiceTest.java | 90 +++++ .../remote/RemoteProgressRelayTest.java | 86 +++++ 24 files changed, 1845 insertions(+), 7 deletions(-) create mode 100644 README_BUILD_PROGRESS.md create mode 100644 client/src/main/java/com/defold/extender/client/ExtenderProgressConsumer.java create mode 100644 client/src/main/java/com/defold/extender/client/ExtenderProgressListener.java create mode 100644 server/src/main/java/com/defold/extender/BuildProgressController.java create mode 100644 server/src/main/java/com/defold/extender/progress/BuildProgressEvent.java create mode 100644 server/src/main/java/com/defold/extender/progress/BuildProgressService.java create mode 100644 server/src/main/java/com/defold/extender/progress/BuildStage.java create mode 100644 server/src/main/java/com/defold/extender/progress/ProgressReporter.java create mode 100644 server/src/main/java/com/defold/extender/remote/RemoteProgressRelay.java create mode 100644 server/src/test/java/com/defold/extender/BuildProgressControllerTest.java create mode 100644 server/src/test/java/com/defold/extender/process/ProcessExecutorTest.java create mode 100644 server/src/test/java/com/defold/extender/progress/BuildProgressServiceTest.java create mode 100644 server/src/test/java/com/defold/extender/remote/RemoteProgressRelayTest.java diff --git a/README.md b/README.md index 0c4ffbc2..8aa1a7b4 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,7 @@ Extender is a build server that builds native extensions of the Defold engine. T * Server description and setup/run instructions - [link](/server/README.md) * Debugging FAQ - [link](/README_DEBUGGING.md) +* Live build progress (SSE) - [link](/README_BUILD_PROGRESS.md) --- diff --git a/README_BUILD_PROGRESS.md b/README_BUILD_PROGRESS.md new file mode 100644 index 00000000..d03a0f62 --- /dev/null +++ b/README_BUILD_PROGRESS.md @@ -0,0 +1,106 @@ +# Live build progress + +The extender streams live build progress over Server-Sent Events (SSE) so clients can show +what a build is doing (downloading the SDK, resolving dependencies, compiling file N of M, +linking, packaging) instead of a silent wait. + +Progress is **advisory**: `/job_status` polling remains the source of truth for build +completion, and terminal progress events are only emitted after the result files +(`build.zip`/`error.txt`) are in place. Everything is backward compatible — old clients +never call the endpoint, and new clients fall back to polling when the server has no +progress support. + +## Endpoint + +``` +GET /job_progress?jobId= +Accept: text/event-stream +``` + +* Live job: streams `progress` events. A snapshot of the current state is sent immediately + on subscribe (early events always precede the first subscriber, because the build is + dispatched before the jobId is returned). +* Reconnects can pass the standard `Last-Event-ID` header; missed events are replayed from + a per-job ring buffer when possible, otherwise a state snapshot is sent. +* Job already finished (result files on disk): a single terminal event, then the stream closes. +* Unknown job, or `extender.progress.enabled: false`: `404`. +* Comment lines (`:ka`) are heartbeats sent every `heartbeat-interval` to keep idle + connections alive through proxies. + +Event payload (JSON, `id:` field = `seq`): + +```json +{ + "jobId": "job1234567890", + "seq": 17, + "ts": 1720512345678, + "stage": "COMPILING", + "detail": "extension1: compiling source files", + "percent": 55, + "extension": "extension1", + "currentFile": 12, + "totalFiles": 34, + "terminal": false +} +``` + +Stages, in pipeline order: `RECEIVED`, `QUEUED`, `SDK`, `DEPENDENCIES`, `MANIFESTS`, +`PLATFORM`, `COMPILING`, `LINKING`, `PACKAGING`, and the terminals `SUCCESS`/`ERROR`. +`REMOTE_BUILDING` is a coarse stage used by a frontend when its remote builder runs an +older server without progress support. `percent` is a 0-100 estimate and never decreases. +`extension`/`currentFile`/`totalFiles` are only present while compiling. + +## Frontend / remote builder setups + +A frontend instance relays the remote builder's progress stream to its own subscribers +under the frontend's jobId, so external clients only ever talk to the frontend. If the +remote builder runs an older server version the frontend degrades to coarse +`REMOTE_BUILDING` ticks. + +If the server sits behind a buffering reverse proxy, response buffering must be disabled +for `/job_progress` (e.g. nginx `proxy_buffering off` or the `X-Accel-Buffering: no` +response header), otherwise events arrive in bursts or not at all. The built-in heartbeats +keep idle timeouts (Jetty and load balancers) from closing quiet streams. + +## Server configuration (`application.yml`) + +```yaml +extender: + progress: + enabled: true # false restores the old behavior exactly + sse-timeout: 1800000 # SseEmitter timeout, ms + heartbeat-interval: 15000 # keepalive comment cadence, ms + event-buffer-size: 256 # per-job replay buffer for Last-Event-ID + max-subscribers-per-job: 8 + registry-ttl: 1200000 # sweep for jobs that died without a terminal event + cleanup-period: 20000 +``` + +## Client API + +`ExtenderClient` gained an overload that reports progress while the existing polling flow +runs unchanged: + +```java +extenderClient.build(platform, sdkVersion, sourceResources, destination, log, + (stage, detail, percent, currentFile, totalFiles) -> { + // called on a background thread; currentFile/totalFiles are -1 outside COMPILING + System.out.printf("[%3d%%] %s %s%n", percent, stage, detail); + }); +``` + +System properties: + +* `com.defold.extender.client.progress-enabled` (default `true`) — set `false` to never + open the progress stream. +* `com.defold.extender.client.progress-reconnect-attempts` (default `5`) — reconnect + attempts for a dropped stream. + +## Trying it with curl + +```sh +JOB=$(curl -s -X POST -F "file=@upload.zip" http://localhost:9000/build_async/x86_64-linux/) +curl -sN "http://localhost:9000/job_progress?jobId=$JOB" +# reconnect mid-build and replay everything after event 10: +curl -sN -H "Last-Event-ID: 10" "http://localhost:9000/job_progress?jobId=$JOB" +``` diff --git a/README_CLIENT.md b/README_CLIENT.md index f6576123..feacda1a 100644 --- a/README_CLIENT.md +++ b/README_CLIENT.md @@ -12,4 +12,7 @@ There is a client part of the Extender code which is used in Bob.jar. $ cp -v ./build/libs/extender-client-0.0.1.jar /com.dynamo.cr/com.dynamo.cr.common/ext/extender-client-0.0.1.jar +The client can report live build progress through an `ExtenderProgressListener` - +see [README_BUILD_PROGRESS.md](/README_BUILD_PROGRESS.md). + diff --git a/client/src/main/java/com/defold/extender/client/ExtenderClient.java b/client/src/main/java/com/defold/extender/client/ExtenderClient.java index 0ea76865..ee8a99eb 100644 --- a/client/src/main/java/com/defold/extender/client/ExtenderClient.java +++ b/client/src/main/java/com/defold/extender/client/ExtenderClient.java @@ -58,6 +58,7 @@ public class ExtenderClient { private ExtenderClientCache cache; private long buildSleepTimeout; private long buildResultWaitTimeout; + private boolean progressEnabled; private List headers; private HttpClient httpClient; @@ -97,6 +98,7 @@ public ExtenderClient(ExtenderClientCache cache, this.cache = cache; this.buildSleepTimeout = Long.parseLong(System.getProperty("com.defold.extender.client.build-sleep-timeout", "5000")); this.buildResultWaitTimeout = Long.parseLong(System.getProperty("com.defold.extender.client.build-wait-timeout", "1200000")); + this.progressEnabled = Boolean.parseBoolean(System.getProperty("com.defold.extender.client.progress-enabled", "true")); this.headers = new ArrayList(); this.httpClient = httpClient; } @@ -238,7 +240,8 @@ static Set getCachedFiles(String json) throws ExtenderClientException { } - private void build_async(String platform, String sdkVersion, HttpEntity entity, File destination, File log) throws ExtenderClientException { + private void build_async(String platform, String sdkVersion, HttpEntity entity, File destination, File log, ExtenderProgressListener progressListener) throws ExtenderClientException { + ExtenderProgressConsumer progressConsumer = null; try { String url = String.format("%s/build_async/%s/%s", extenderBaseUrl, platform, sdkVersion); HttpPost request = createPostRequest(url); @@ -254,6 +257,15 @@ private void build_async(String platform, String sdkVersion, HttpEntity entity, String jobId = EntityUtils.toString(response.getEntity()); String traceId = response.getFirstHeader(TRACE_ID_HEADER_NAME).getValue(); log("Async build request was accepted as job %s (traceId: %s)", jobId, traceId == null ? "null" : traceId); + if (progressListener != null && progressEnabled) { + // advisory SSE progress stream; the poll loop below stays + // the sole authority on build completion + progressConsumer = new ExtenderProgressConsumer(httpClient, extenderBaseUrl, jobId, + this::createGetRequest, progressListener); + Thread progressThread = new Thread(progressConsumer, "extender-progress-" + jobId); + progressThread.setDaemon(true); + progressThread.start(); + } long currentTime = System.currentTimeMillis(); Integer jobStatus = 0; Thread.sleep(buildSleepTimeout); @@ -312,6 +324,10 @@ private void build_async(String platform, String sdkVersion, HttpEntity entity, } catch (Exception e) { throw new ExtenderClientException("Failed to communicate with Extender service.", e); + } finally { + if (progressConsumer != null) { + progressConsumer.stop(); + } } } @@ -362,6 +378,18 @@ HttpEntity createBuildRequestPayload(List sourceResources) thr * @throws ExtenderClientException */ public void build(String platform, String sdkVersion, List sourceResources, File destination, File log) throws ExtenderClientException { + build(platform, sdkVersion, sourceResources, destination, log, null); + } + + /** + * Builds a new engine and reports live build progress. + * + * @param progressListener Receives progress updates on a background thread + * while the build runs, or null. Progress is advisory: + * it may stop arriving (old server, dropped connection) + * while the build continues. + */ + public void build(String platform, String sdkVersion, List sourceResources, File destination, File log, ExtenderProgressListener progressListener) throws ExtenderClientException { String cacheKey = cache.calcKey(platform, sdkVersion, sourceResources); boolean isCached = cache.isCached(platform, cacheKey); if (isCached) { @@ -370,7 +398,7 @@ public void build(String platform, String sdkVersion, List sou } HttpEntity payload = createBuildRequestPayload(sourceResources); - build_async(platform, sdkVersion, payload, destination, log); + build_async(platform, sdkVersion, payload, destination, log, progressListener); // Store the new build cache.put(platform, cacheKey, destination); diff --git a/client/src/main/java/com/defold/extender/client/ExtenderProgressConsumer.java b/client/src/main/java/com/defold/extender/client/ExtenderProgressConsumer.java new file mode 100644 index 00000000..e60a31d8 --- /dev/null +++ b/client/src/main/java/com/defold/extender/client/ExtenderProgressConsumer.java @@ -0,0 +1,167 @@ +package com.defold.extender.client; + +import org.apache.http.HttpResponse; +import org.apache.http.HttpStatus; +import org.apache.http.client.HttpClient; +import org.apache.http.client.config.RequestConfig; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.util.EntityUtils; +import org.json.simple.JSONObject; +import org.json.simple.parser.JSONParser; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * Consumes the server's /job_progress SSE stream on a background thread + * and forwards events to an ExtenderProgressListener. + * + * Strictly advisory: any failure (404 from an old server, dropped + * connection, malformed data) is swallowed after bounded reconnect + * attempts. The poll loop in ExtenderClient.build_async remains the sole + * authority on build completion and calls stop() when the build is done. + */ +class ExtenderProgressConsumer implements Runnable { + private static final Logger logger = Logger.getLogger(ExtenderProgressConsumer.class.getName()); + + /** Creates GET requests carrying the client's auth and custom headers. */ + interface GetRequestFactory { + HttpGet create(String url) throws IOException; + } + + private static final long RECONNECT_BACKOFF_MS = 2000; + + private final HttpClient httpClient; + private final String jobProgressUrl; + private final GetRequestFactory requestFactory; + private final ExtenderProgressListener listener; + private final int maxReconnectAttempts; + + private volatile boolean stopped = false; + private volatile HttpGet currentRequest = null; + private long lastEventId = -1; + + ExtenderProgressConsumer(HttpClient httpClient, String extenderBaseUrl, String jobId, + GetRequestFactory requestFactory, ExtenderProgressListener listener) { + this.httpClient = httpClient; + this.jobProgressUrl = String.format("%s/job_progress?jobId=%s", extenderBaseUrl, jobId); + this.requestFactory = requestFactory; + this.listener = listener; + this.maxReconnectAttempts = Integer.parseInt( + System.getProperty("com.defold.extender.client.progress-reconnect-attempts", "5")); + } + + /** Stops the consumer and unblocks the stream read. Safe to call more than once. */ + void stop() { + stopped = true; + HttpGet request = currentRequest; + if (request != null) { + request.abort(); + } + } + + @Override + public void run() { + try { + int attempts = 0; + while (!stopped && attempts < maxReconnectAttempts) { + attempts++; + try { + if (stream()) { + return; // unsupported by server or terminal event seen + } + } catch (IOException e) { + if (stopped) { + return; + } + logger.log(Level.FINE, "Build progress stream dropped, reconnecting: " + e.getMessage()); + } + Thread.sleep(RECONNECT_BACKOFF_MS); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (Exception e) { + // progress must never break the build + logger.log(Level.FINE, "Build progress consumer stopped: " + e.getMessage()); + } + } + + /** + * Opens the SSE stream and forwards events until it ends. + * Returns true when the consumer is done for good (server has no + * progress support, or a terminal event arrived); false to reconnect. + */ + private boolean stream() throws IOException { + HttpGet request = requestFactory.create(jobProgressUrl); + request.setHeader("Accept", "text/event-stream"); + if (lastEventId >= 0) { + request.setHeader("Last-Event-ID", Long.toString(lastEventId)); + } + // the stream stays open for the whole build: disable the socket + // read timeout for this request; stop() aborts it when the build is done + request.setConfig(RequestConfig.custom().setSocketTimeout(0).build()); + currentRequest = request; + try { + HttpResponse response = httpClient.execute(request); + if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) { + // old server or progress disabled; polling still reports completion + EntityUtils.consumeQuietly(response.getEntity()); + return true; + } + try (BufferedReader reader = new BufferedReader( + new InputStreamReader(response.getEntity().getContent(), StandardCharsets.UTF_8))) { + String eventId = null; + StringBuilder data = new StringBuilder(); + String line; + while (!stopped && (line = reader.readLine()) != null) { + if (line.isEmpty()) { + // end of one SSE event + if (data.length() > 0 && dispatch(eventId, data.toString())) { + return true; // terminal event + } + eventId = null; + data.setLength(0); + } else if (line.startsWith("id:")) { + eventId = line.substring(3).trim(); + } else if (line.startsWith("data:")) { + data.append(line.substring(5).trim()); + } + // "event:" names and ":" comments (heartbeats) are ignored + } + } + return stopped; + } finally { + currentRequest = null; + } + } + + /** Forwards one event to the listener. Returns true for terminal events. */ + private boolean dispatch(String eventId, String data) { + boolean terminal = false; + try { + JSONObject json = (JSONObject) new JSONParser().parse(data); + if (eventId != null) { + lastEventId = Long.parseLong(eventId); + } + String stage = (String) json.get("stage"); + String detail = (String) json.get("detail"); + Number percent = (Number) json.get("percent"); + Number currentFile = (Number) json.get("currentFile"); + Number totalFiles = (Number) json.get("totalFiles"); + Boolean isTerminal = (Boolean) json.get("terminal"); + terminal = isTerminal != null && isTerminal; + listener.onProgress(stage, detail, + percent != null ? percent.intValue() : 0, + currentFile != null ? currentFile.intValue() : -1, + totalFiles != null ? totalFiles.intValue() : -1); + } catch (Exception e) { + // a malformed event or a listener bug must not kill the stream + logger.log(Level.FINE, "Ignoring bad progress event: " + e.getMessage()); + } + return terminal; + } +} diff --git a/client/src/main/java/com/defold/extender/client/ExtenderProgressListener.java b/client/src/main/java/com/defold/extender/client/ExtenderProgressListener.java new file mode 100644 index 00000000..5b30bd81 --- /dev/null +++ b/client/src/main/java/com/defold/extender/client/ExtenderProgressListener.java @@ -0,0 +1,24 @@ +package com.defold.extender.client; + +/** + * Receives live build-progress updates while ExtenderClient.build(...) is + * waiting for the server to finish a build. + * + * Progress is advisory: it may stop arriving at any time (old server, + * dropped connection) while the build itself keeps running. Completion is + * always determined by the build call returning or throwing. + * + * Callbacks are invoked on a background thread, never on the thread that + * called build(...). + */ +public interface ExtenderProgressListener { + /** + * @param stage Current build stage, e.g. "SDK", "DEPENDENCIES", + * "COMPILING", "LINKING", "PACKAGING", "SUCCESS", "ERROR" + * @param detail Human-readable detail line, e.g. the extension being compiled. May be null. + * @param percent Overall progress estimate 0-100, never decreasing. + * @param currentFile Files compiled so far for the current extension, or -1 when not compiling. + * @param totalFiles Total files to compile for the current extension, or -1 when not compiling. + */ + void onProgress(String stage, String detail, int percent, int currentFile, int totalFiles); +} diff --git a/client/src/test/java/com/defold/extender/client/ExtenderClientTest.java b/client/src/test/java/com/defold/extender/client/ExtenderClientTest.java index 2da66957..8e55cfd4 100644 --- a/client/src/test/java/com/defold/extender/client/ExtenderClientTest.java +++ b/client/src/test/java/com/defold/extender/client/ExtenderClientTest.java @@ -36,12 +36,23 @@ import java.io.IOException; import java.lang.reflect.Field; import java.lang.reflect.InvocationTargetException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.util.ArrayList; +import java.util.Collections; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.stream.Stream; import java.util.zip.ZipEntry; import java.util.zip.ZipInputStream; +import org.apache.http.Header; +import org.apache.http.HttpResponse; +import org.apache.http.message.BasicHeader; +import org.apache.http.message.BasicStatusLine; +import org.apache.http.ProtocolVersion; + public class ExtenderClientTest extends Mockito { @BeforeAll public static void beforeClass() { @@ -130,6 +141,136 @@ public void testClientHandleHTTPError() throws ClientProtocolException, IOExcept } + private HttpResponse mockResponse(int statusCode, byte[] body, Header... headers) { + CloseableHttpResponse response = Mockito.mock(CloseableHttpResponse.class); + when(response.getStatusLine()).thenReturn( + new BasicStatusLine(new ProtocolVersion("HTTP", 1, 1), statusCode, "")); + when(response.getEntity()).thenReturn( + EntityBuilder.create().setStream(new ByteArrayInputStream(body)).build()); + for (Header header : headers) { + when(response.getFirstHeader(header.getName())).thenReturn(header); + } + return response; + } + + // mocks a full successful async build: /query 404, /build_async accepted, + // /job_status done, /job_result zip bytes, /job_progress as given + private DefaultHttpClient mockBuildHttpClient(int progressStatusCode, String progressSseBody) throws IOException { + DefaultHttpClient httpClient = Mockito.mock(DefaultHttpClient.class); + when(httpClient.execute(Mockito.any(HttpPost.class))).thenAnswer(invocation -> { + HttpPost request = invocation.getArgument(0); + if (request.getURI().toString().contains("/build_async")) { + return mockResponse(200, "job123".getBytes(), new BasicHeader("X-TraceId", "trace123")); + } + return mockResponse(404, "no cache".getBytes()); // /query: caching unsupported + }); + when(httpClient.execute(Mockito.any(HttpGet.class))).thenAnswer(invocation -> { + HttpGet request = invocation.getArgument(0); + String uri = request.getURI().toString(); + if (uri.contains("/job_progress")) { + return mockResponse(progressStatusCode, progressSseBody.getBytes(StandardCharsets.UTF_8)); + } + if (uri.contains("/job_status")) { + return mockResponse(200, "1".getBytes()); + } + if (uri.contains("/job_result")) { + return mockResponse(200, "zipcontent".getBytes()); + } + return mockResponse(404, new byte[0]); + }); + return httpClient; + } + + private static List progressTestSources(File dir) throws IOException { + File src = new File(dir, "source.cpp"); + Files.write(src.toPath(), "int main() {}".getBytes()); + src.deleteOnExit(); + List sources = new ArrayList<>(); + sources.add(new FileExtenderResource(src)); + return sources; + } + + @Test + public void testBuildWithProgressListener() throws Exception { + String sse = "id:1\n" + + "data:{\"jobId\":\"job123\",\"seq\":1,\"ts\":1,\"stage\":\"SDK\",\"detail\":\"Downloading\",\"percent\":1,\"terminal\":false}\n" + + "\n" + + ":ka\n" + + "\n" + + "id:2\n" + + "data:{\"jobId\":\"job123\",\"seq\":2,\"ts\":2,\"stage\":\"COMPILING\",\"detail\":\"ext1\",\"percent\":55,\"extension\":\"ext1\",\"currentFile\":12,\"totalFiles\":34,\"terminal\":false}\n" + + "\n" + + "id:3\n" + + "data:{\"jobId\":\"job123\",\"seq\":3,\"ts\":3,\"stage\":\"SUCCESS\",\"detail\":\"done\",\"percent\":100,\"terminal\":true}\n" + + "\n"; + + String oldSleepTimeout = System.setProperty("com.defold.extender.client.build-sleep-timeout", "10"); + try { + File cacheDir = Files.createTempDirectory("progress-test-cache").toFile(); + cacheDir.deleteOnExit(); + File destination = new File(cacheDir, "build.zip"); + File log = new File(cacheDir, "build.log"); + + ExtenderClient extenderClient = new ExtenderClient(new ExtenderClientCache(cacheDir), + mockBuildHttpClient(200, sse), "http://localhost"); + + List events = Collections.synchronizedList(new ArrayList<>()); + CountDownLatch terminalSeen = new CountDownLatch(1); + ExtenderProgressListener listener = (stage, detail, percent, currentFile, totalFiles) -> { + events.add(String.format("%s:%d:%d/%d", stage, percent, currentFile, totalFiles)); + if ("SUCCESS".equals(stage) || "ERROR".equals(stage)) { + terminalSeen.countDown(); + } + }; + + extenderClient.build("x86_64-linux", "testsdk1", progressTestSources(cacheDir), destination, log, listener); + + assertTrue(destination.exists()); + assertEquals("zipcontent", new String(Files.readAllBytes(destination.toPath()))); + assertTrue(terminalSeen.await(5, TimeUnit.SECONDS), "listener never saw the terminal event"); + assertTrue(events.contains("SDK:1:-1/-1"), "events: " + events); + assertTrue(events.contains("COMPILING:55:12/34"), "events: " + events); + assertTrue(events.contains("SUCCESS:100:-1/-1"), "events: " + events); + } finally { + restoreProperty("com.defold.extender.client.build-sleep-timeout", oldSleepTimeout); + } + } + + @Test + public void testBuildSucceedsWhenServerHasNoProgressSupport() throws Exception { + String oldSleepTimeout = System.setProperty("com.defold.extender.client.build-sleep-timeout", "10"); + try { + File cacheDir = Files.createTempDirectory("progress-test-cache-404").toFile(); + cacheDir.deleteOnExit(); + File destination = new File(cacheDir, "build.zip"); + File log = new File(cacheDir, "build.log"); + + // old server: /job_progress answers 404; the build must be unaffected + ExtenderClient extenderClient = new ExtenderClient(new ExtenderClientCache(cacheDir), + mockBuildHttpClient(404, "not found"), "http://localhost"); + + List events = Collections.synchronizedList(new ArrayList<>()); + ExtenderProgressListener listener = (stage, detail, percent, currentFile, totalFiles) -> + events.add(stage); + + extenderClient.build("x86_64-linux", "testsdk2", progressTestSources(cacheDir), destination, log, listener); + + assertTrue(destination.exists()); + assertEquals("zipcontent", new String(Files.readAllBytes(destination.toPath()))); + assertTrue(events.isEmpty(), "no progress events expected from an old server, got: " + events); + } finally { + restoreProperty("com.defold.extender.client.build-sleep-timeout", oldSleepTimeout); + } + } + + private static void restoreProperty(String name, String oldValue) { + if (oldValue == null) { + System.clearProperty(name); + } else { + System.setProperty(name, oldValue); + } + } + private static Stream uploadData() { return Stream.of( Arguments.of("{\"files\":[{\"cached\":true,\"path\":\"build/a\"},{\"cached\":true,\"path\":\"build/b\"},{\"cached\":false,\"path\":\"build/c\"}]}", List.of("build/c")), diff --git a/server/src/main/java/com/defold/extender/AsyncBuilder.java b/server/src/main/java/com/defold/extender/AsyncBuilder.java index bc0bbd29..39aeb2f2 100644 --- a/server/src/main/java/com/defold/extender/AsyncBuilder.java +++ b/server/src/main/java/com/defold/extender/AsyncBuilder.java @@ -13,6 +13,9 @@ import com.defold.extender.log.Markers; import com.defold.extender.metrics.MetricsWriter; +import com.defold.extender.progress.BuildProgressService; +import com.defold.extender.progress.BuildStage; +import com.defold.extender.progress.ProgressReporter; import com.defold.extender.services.DefoldSdkService; import com.defold.extender.services.GradleService; import com.defold.extender.services.cocoapods.CocoaPodsService; @@ -35,6 +38,7 @@ public class AsyncBuilder { private DefoldSdkService defoldSdkService; private GradleService gradleService; private CocoaPodsService cocoaPodsService; + private BuildProgressService buildProgressService; private File jobResultLocation; private long resultLifetime; private boolean keepJobDirectory = false; @@ -42,11 +46,13 @@ public class AsyncBuilder { public AsyncBuilder(DefoldSdkService defoldSdkService, GradleService gradleService, Optional cocoaPodsService, + BuildProgressService buildProgressService, @Value("${extender.job-result.location}") String jobResultLocation, @Value("${extender.job-result.lifetime:1200000}") long jobResultLifetime) { this.defoldSdkService = defoldSdkService; this.gradleService = gradleService; cocoaPodsService.ifPresent(val -> { this.cocoaPodsService = val; }); + this.buildProgressService = buildProgressService; this.jobResultLocation = new File(jobResultLocation); this.keepJobDirectory = System.getenv("DM_DEBUG_KEEP_JOB_FOLDER") != null || System.getenv("DM_DEBUG_JOB_FOLDER") != null; this.resultLifetime = jobResultLifetime; @@ -91,10 +97,12 @@ public void asyncBuildEngine(MetricsWriter metricsWriter, String platform, Strin resultDir.mkdir(); Extender extender = null; Boolean isSuccefull = true; + ProgressReporter progressReporter = buildProgressService.reporterFor(jobName); try { LOGGER.info("Building engine locally"); // Get SDK + progressReporter.stage(BuildStage.SDK, "Downloading Defold SDK " + sdkVersion); try (DefoldSdk sdk = defoldSdkService.getSdk(sdkVersion)) { metricsWriter.measureSdkDownload(sdkVersion); @@ -105,16 +113,19 @@ public void asyncBuildEngine(MetricsWriter metricsWriter, String platform, Strin .setUploadDirectory(uploadDirectory) .setBuildDirectory(buildDirectory) .setMetricsWriter(metricsWriter) + .setProgressReporter(progressReporter) .build(); // Resolve Gradle dependencies if (platform.contains("android")) { + progressReporter.stage(BuildStage.DEPENDENCIES, "Resolving Gradle dependencies"); extender.resolve(gradleService); metricsWriter.measureGradleDownload(); } // Resolve CocoaPods dependencies if (ExtenderUtil.isAppleTarget(platform)) { + progressReporter.stage(BuildStage.DEPENDENCIES, "Resolving CocoaPods dependencies"); extender.resolve(cocoaPodsService); metricsWriter.measureCocoaPodsInstallation(); } @@ -124,6 +135,7 @@ public void asyncBuildEngine(MetricsWriter metricsWriter, String platform, Strin metricsWriter.measureEngineBuild(platform); // Zip files + progressReporter.stage(BuildStage.PACKAGING, "Packaging build results"); String zipFilename = jobDirectory.getAbsolutePath() + File.separator + BuilderConstants.BUILD_RESULT_FILENAME; File zipFile = ZipUtils.zip(extender.getOutputFiles(), buildDirectory, zipFilename); metricsWriter.measureZipFiles(zipFile); @@ -133,6 +145,10 @@ public void asyncBuildEngine(MetricsWriter metricsWriter, String platform, Strin File targetResult = new File(resultDir, BuilderConstants.BUILD_RESULT_FILENAME); FileUtils.copyFile(zipFile, tmpResult); Files.move(tmpResult.toPath(), targetResult.toPath(), StandardCopyOption.ATOMIC_MOVE); + // terminal event only after the result file is in place, so + // /job_status and /job_result are already consistent for + // clients reacting to it + progressReporter.terminal(true, "Build succeeded"); } } catch(EofException e) { File errorFile = new File(resultDir, BuilderConstants.BUILD_ERROR_FILENAME); @@ -140,12 +156,14 @@ public void asyncBuildEngine(MetricsWriter metricsWriter, String platform, Strin writeExceptionToFile(e, errorFile); LOGGER.error(Markers.SERVER_ERROR, "Client closed connection prematurely, build aborted", e); isSuccefull = false; + progressReporter.terminal(false, "Build aborted: client closed connection"); } catch(Exception e) { File errorFile = new File(resultDir, BuilderConstants.BUILD_ERROR_FILENAME); writeExtenderLogsToFile(extender, errorFile); writeExceptionToFile(e, errorFile); LOGGER.error(String.format("Exception while building or sending response - SDK: %s", sdkVersion), e); isSuccefull = false; + progressReporter.terminal(false, "Build failed: " + e.getMessage()); } finally { metricsWriter.measureCounterBuild(platform, sdkVersion, "async", isSuccefull); diff --git a/server/src/main/java/com/defold/extender/BuildProgressController.java b/server/src/main/java/com/defold/extender/BuildProgressController.java new file mode 100644 index 00000000..365b7955 --- /dev/null +++ b/server/src/main/java/com/defold/extender/BuildProgressController.java @@ -0,0 +1,88 @@ +package com.defold.extender; + +import com.defold.extender.progress.BuildProgressService; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestHeader; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.io.File; +import java.io.IOException; + +/** + * Streams live build progress as Server-Sent Events. Advisory only: + * clients must keep polling /job_status for completion. Old clients never + * call this endpoint; new clients treat a 404 (old server or feature + * disabled) as "no progress available" and fall back to polling. + */ +@RestController +public class BuildProgressController { + private static final Logger LOGGER = LoggerFactory.getLogger(BuildProgressController.class); + + private final BuildProgressService progressService; + private final File jobResultLocation; + + public BuildProgressController(BuildProgressService progressService, + @Value("${extender.job-result.location}") String jobResultLocation) { + this.progressService = progressService; + this.jobResultLocation = new File(jobResultLocation); + } + + @GetMapping(path = "/job_progress", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public ResponseEntity jobProgress(@RequestParam(name = "jobId") String jobId, + @RequestHeader(name = "Last-Event-ID", required = false) String lastEventIdHeader) + throws IOException { + if (!progressService.isEnabled()) { + return ResponseEntity.notFound().build(); + } + + SseEmitter emitter; + try { + emitter = progressService.subscribe(jobId, parseLastEventId(lastEventIdHeader)); + } catch (IllegalStateException e) { + LOGGER.warn(e.getMessage()); + return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS).build(); + } + if (emitter != null) { + return ResponseEntity.ok(emitter); + } + + // Not a live job: if the result files are already on disk (subscribed + // after the build finished, or the server restarted), send a single + // terminal event so the subscriber is not left hanging. + File jobResultDir; + try { + jobResultDir = SandboxedPath.resolve(jobResultLocation, jobId); + } catch (ExtenderException e) { + return ResponseEntity.notFound().build(); + } + if (jobResultDir.exists()) { + if (new File(jobResultDir, BuilderConstants.BUILD_RESULT_FILENAME).exists()) { + return ResponseEntity.ok(progressService.finishedJobEmitter(jobId, true)); + } + if (new File(jobResultDir, BuilderConstants.BUILD_ERROR_FILENAME).exists()) { + return ResponseEntity.ok(progressService.finishedJobEmitter(jobId, false)); + } + } + return ResponseEntity.notFound().build(); + } + + private static Long parseLastEventId(String header) { + if (header == null) { + return null; + } + try { + return Long.valueOf(header.trim()); + } catch (NumberFormatException e) { + return null; + } + } +} diff --git a/server/src/main/java/com/defold/extender/Extender.java b/server/src/main/java/com/defold/extender/Extender.java index 8b701c85..e3042d93 100644 --- a/server/src/main/java/com/defold/extender/Extender.java +++ b/server/src/main/java/com/defold/extender/Extender.java @@ -51,6 +51,8 @@ import com.defold.extender.metrics.MetricsWriter; import com.defold.extender.process.ProcessExecutor; import com.defold.extender.process.ProcessUtils; +import com.defold.extender.progress.BuildStage; +import com.defold.extender.progress.ProgressReporter; class Extender { private static final Logger LOGGER = LoggerFactory.getLogger(Extender.class); @@ -62,6 +64,7 @@ class Extender { private final PlatformConfig platformAppConfig; // "common", platform, arch-platform from game.appmanifest private final TemplateExecutor templateExecutor = new TemplateExecutor(); private final ProcessExecutor processExecutor = new ProcessExecutor(); + private final ProgressReporter progressReporter; private MetricsWriter metricsWriter; // context flags private Boolean needsCSLibraries = false; @@ -128,6 +131,7 @@ static public class Builder { File uploadDirectory; Map env = new HashMap(); MetricsWriter metricsWriter; + ProgressReporter progressReporter = ProgressReporter.NOOP; public Builder() { } @@ -166,6 +170,11 @@ public Builder setMetricsWriter(MetricsWriter writer) { return this; } + public Builder setProgressReporter(ProgressReporter progressReporter) { + this.progressReporter = progressReporter; + return this; + } + public Extender build() throws IOException, ExtenderException { return new Extender(this); } @@ -173,6 +182,7 @@ public Extender build() throws IOException, ExtenderException { private Extender(Builder builder) throws IOException, ExtenderException { this.metricsWriter = builder.metricsWriter; + this.progressReporter = builder.progressReporter != null ? builder.progressReporter : ProgressReporter.NOOP; this.gradlePackages = new ArrayList<>(); this.outputFiles = new ArrayList<>(); @@ -815,7 +825,9 @@ else if (platformConfig.zigSourceRe != null && ExtenderUtil.matchesFile(src, pla } objs.add(ExtenderUtil.getRelativePath(buildState.jobDir, o)); } - ProcessExecutor.executeCommands(processExecutor, commands); // in parallel + final String extensionName = (String)manifestContext.get("extension_name"); + progressReporter.compileBatchBegin(extensionName, commands.size()); + ProcessExecutor.executeCommands(processExecutor, commands, () -> progressReporter.fileCompiled(extensionName)); // in parallel return objs; } @@ -917,7 +929,8 @@ private List compilePodSourceFiles(PodBuildSpec pod, Map compileSwiftCommands.set(i, cmd); } // ************************************************************************************************************ - ProcessExecutor.executeCommands(processExecutor, compileSwiftCommands); // in parallel + progressReporter.compileBatchBegin(pod.name, compileSwiftCommands.size()); + ProcessExecutor.executeCommands(processExecutor, compileSwiftCommands, () -> progressReporter.fileCompiled(pod.name)); // in parallel generateSwiftCompatabilityHeaders(pod, resolvedPods.getCurrentPodsDirectory()); } @@ -944,7 +957,8 @@ else if (extension.equals("mm")) { objs.add(objPath); } LOGGER.info("compiling {} source files", commands.size()); - ProcessExecutor.executeCommands(processExecutor, commands); // in parallel + progressReporter.compileBatchBegin(pod.name, commands.size()); + ProcessExecutor.executeCommands(processExecutor, commands, () -> progressReporter.fileCompiled(pod.name)); // in parallel return objs; } @@ -2384,6 +2398,7 @@ private List buildEngine() throws ExtenderException { List outputFiles = new ArrayList<>(); try { + progressReporter.stage(BuildStage.COMPILING, "Building pods"); outputFiles.addAll(buildPods()); // An easy way to disable building an extension, is if the symbol name is @@ -2402,6 +2417,7 @@ private List buildEngine() throws ExtenderException { Map extensionContext = manifestConfigs.get(extensionSymbol); File manifest = manifestFiles.get(extensionSymbol); + progressReporter.stage(BuildStage.COMPILING, extensionSymbol); // TODO: Thread this step outputFiles.addAll(buildExtension(manifest, extensionContext)); } @@ -2422,6 +2438,7 @@ private List buildEngine() throws ExtenderException { } Map mergedAppContextWithPods = ExtenderUtil.mergeContexts(mergedAppContext, podAppContext); + progressReporter.stage(BuildStage.LINKING, "Linking engine"); outputFiles.addAll(linkEngine(symbols, mergedAppContextWithPods, resourceFile)); metricsWriter.measureBuildTarget("engine"); @@ -2818,19 +2835,23 @@ void resolve(CocoaPodsService cocoaPodsService) throws ExtenderException { } void build() throws ExtenderException { + progressReporter.stage(BuildStage.MANIFESTS, "Building manifests"); outputFiles.addAll(buildManifests(buildState.fullPlatform)); if (shouldBuildLibrary()) { + progressReporter.stage(BuildStage.COMPILING, "Building libraries"); outputFiles.addAll(buildLibraries()); } else { // TODO: Thread this step if (ExtenderUtil.isAndroidTarget(buildState.fullPlatform)) { + progressReporter.stage(BuildStage.PLATFORM, "Building Android resources and code"); outputFiles.addAll(buildAndroid(buildState.fullPlatform)); } else if (ExtenderUtil.isAppleTarget(buildState.fullPlatform)) { + progressReporter.stage(BuildStage.PLATFORM, "Building Apple platform files"); outputFiles.addAll(buildApple(buildState.fullPlatform)); } diff --git a/server/src/main/java/com/defold/extender/ExtenderController.java b/server/src/main/java/com/defold/extender/ExtenderController.java index 8320a957..de5c1782 100644 --- a/server/src/main/java/com/defold/extender/ExtenderController.java +++ b/server/src/main/java/com/defold/extender/ExtenderController.java @@ -5,6 +5,9 @@ import com.defold.extender.remote.RemoteInstanceConfig; import com.defold.extender.log.Markers; import com.defold.extender.metrics.MetricsWriter; +import com.defold.extender.progress.BuildProgressService; +import com.defold.extender.progress.BuildStage; +import com.defold.extender.progress.ProgressReporter; import com.defold.extender.services.DefoldSdkService; import com.defold.extender.services.DataCacheService; import com.defold.extender.services.HealthReporterService; @@ -68,6 +71,7 @@ public enum InstanceType { private final UserUpdateService userUpdateService; private final AsyncBuilder asyncBuilder; private final HealthReporterService healthReporter; + private final BuildProgressService buildProgressService; private final RemoteEngineBuilder remoteEngineBuilder; private Map remoteBuilderPlatformMappings; @@ -108,6 +112,7 @@ public ExtenderController(DefoldSdkService defoldSdkService, RemoteEngineBuilder remoteEngineBuilder, RemoteHostConfiguration remoteHostConfiguration, HealthReporterService healthReporter, + BuildProgressService buildProgressService, @Value("${extender.remote-builder.enabled}") boolean remoteBuilderEnabled, @Value("${spring.servlet.multipart.max-request-size}") String maxPackageSize, @Value("${extender.job-result.location}") String jobResultLocation) { @@ -116,6 +121,7 @@ public ExtenderController(DefoldSdkService defoldSdkService, this.meterRegistry = meterRegistry; this.userUpdateService = userUpdateService; this.healthReporter = healthReporter; + this.buildProgressService = buildProgressService; this.remoteEngineBuilder = remoteEngineBuilder; this.remoteBuilderEnabled = remoteBuilderEnabled; @@ -215,7 +221,13 @@ public void buildEngineAsync(HttpServletRequest _request, DataCacheService.DataCacheServiceInfo uploadResultInfo = dataCacheService.cacheFiles(uploadDirectory); metricsWriter.measureCacheUpload(uploadResultInfo.cachedFileSize.longValue(), uploadResultInfo.cachedFileCount.intValue()); + // Register progress tracking before the async dispatch so the + // reporter is in place when the build starts on a worker thread. + ProgressReporter progressReporter = buildProgressService.register(jobDirectory.getName()); + progressReporter.stage(BuildStage.RECEIVED, "Build request received"); + if (instanceType.equals(InstanceType.BUILDER_ONLY)) { + progressReporter.stage(BuildStage.QUEUED, "Build queued"); asyncBuilder.asyncBuildEngine(metricsWriter, platform, sdkVersion, jobDirectory, uploadDirectory, buildDirectory); } else { String[] buildEnvDescription = null; @@ -235,8 +247,10 @@ public void buildEngineAsync(HttpServletRequest _request, if (remoteBuilderEnabled && buildEnvDescription != null && isRemotePlatform(buildEnvDescription[0], buildEnvDescription[1])) { LOGGER.info("Building engine on remote builder"); RemoteInstanceConfig remoteInstanceConfig = getRemoteBuilderConfig(buildEnvDescription[0], buildEnvDescription[1]); + progressReporter.stage(BuildStage.QUEUED, "Build queued on remote builder"); this.remoteEngineBuilder.buildAsync(remoteInstanceConfig, uploadDirectory, platform, sdkVersion, jobDirectory, metricsWriter); } else if (instanceType.equals(InstanceType.MIXED)) { + progressReporter.stage(BuildStage.QUEUED, "Build queued"); asyncBuilder.asyncBuildEngine(metricsWriter, platform, sdkVersion, jobDirectory, uploadDirectory, buildDirectory); } else { // no remote builder was found and current instance can't build @@ -266,6 +280,10 @@ public void buildEngineAsync(HttpServletRequest _request, if (DM_DEBUG_JOB_FOLDER != null) { deleteDirectory = false; } + if (!isBuildStarted) { + // the build never dispatched; drop the progress entry + buildProgressService.remove(jobDirectory.getName()); + } // Delete temporary upload directory if (deleteDirectory && !isBuildStarted) { if (!FileUtils.deleteQuietly(jobDirectory)) { diff --git a/server/src/main/java/com/defold/extender/process/ProcessExecutor.java b/server/src/main/java/com/defold/extender/process/ProcessExecutor.java index 6313ce28..b06160e7 100644 --- a/server/src/main/java/com/defold/extender/process/ProcessExecutor.java +++ b/server/src/main/java/com/defold/extender/process/ProcessExecutor.java @@ -135,11 +135,19 @@ public void putLog(String msg) { } public static void executeCommands(ProcessExecutor processExecutor, List commands) throws IOException, InterruptedException, ExtenderException { + executeCommands(processExecutor, commands, null); + } + + // onCommandComplete is invoked concurrently from the pool threads, once per successful command + public static void executeCommands(ProcessExecutor processExecutor, List commands, Runnable onCommandComplete) throws IOException, InterruptedException, ExtenderException { ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); List> callables = new ArrayList<>(); for (String command : commands) { callables.add(() -> { processExecutor.execute(command); + if (onCommandComplete != null) { + onCommandComplete.run(); + } return null; }); } diff --git a/server/src/main/java/com/defold/extender/progress/BuildProgressEvent.java b/server/src/main/java/com/defold/extender/progress/BuildProgressEvent.java new file mode 100644 index 00000000..165cf9d4 --- /dev/null +++ b/server/src/main/java/com/defold/extender/progress/BuildProgressEvent.java @@ -0,0 +1,47 @@ +package com.defold.extender.progress; + +import com.fasterxml.jackson.annotation.JsonInclude; + +/** + * A single progress update for an async build job. Serialized to JSON and + * pushed to subscribers as an SSE "progress" event, with {@link #getSeq()} + * as the SSE event id (used for Last-Event-ID replay). + */ +@JsonInclude(JsonInclude.Include.NON_NULL) +public class BuildProgressEvent { + private final String jobId; + private final long seq; + private final long ts; + private final BuildStage stage; + private final String detail; + private final int percent; + private final String extension; + private final Integer currentFile; + private final Integer totalFiles; + private final boolean terminal; + + public BuildProgressEvent(String jobId, long seq, long ts, BuildStage stage, String detail, + int percent, String extension, Integer currentFile, Integer totalFiles) { + this.jobId = jobId; + this.seq = seq; + this.ts = ts; + this.stage = stage; + this.detail = detail; + this.percent = percent; + this.extension = extension; + this.currentFile = currentFile; + this.totalFiles = totalFiles; + this.terminal = stage.isTerminal(); + } + + public String getJobId() { return jobId; } + public long getSeq() { return seq; } + public long getTs() { return ts; } + public BuildStage getStage() { return stage; } + public String getDetail() { return detail; } + public int getPercent() { return percent; } + public String getExtension() { return extension; } + public Integer getCurrentFile() { return currentFile; } + public Integer getTotalFiles() { return totalFiles; } + public boolean isTerminal() { return terminal; } +} diff --git a/server/src/main/java/com/defold/extender/progress/BuildProgressService.java b/server/src/main/java/com/defold/extender/progress/BuildProgressService.java new file mode 100644 index 00000000..fadb5d63 --- /dev/null +++ b/server/src/main/java/com/defold/extender/progress/BuildProgressService.java @@ -0,0 +1,359 @@ +package com.defold.extender.progress; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.http.MediaType; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.io.IOException; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +/** + * In-memory registry of per-job build progress. Producers (controller, + * AsyncBuilder, Extender) report through the {@link ProgressReporter} + * returned by {@link #register}/{@link #reporterFor}; consumers subscribe + * with {@link #subscribe} and receive JSON events over SSE. + * + * Entries live from register() until the terminal event (or the TTL sweep + * for builds that died without one). SSE is advisory: job completion is + * still determined by /job_status, and terminal events are only emitted + * after the result files are in place. + */ +@Service +public class BuildProgressService { + private static final Logger LOGGER = LoggerFactory.getLogger(BuildProgressService.class); + + private static final int MAX_DETAIL_LENGTH = 512; + public static final String EVENT_NAME = "progress"; + + private final boolean enabled; + private final long sseTimeout; + private final int eventBufferSize; + private final int maxSubscribersPerJob; + private final long registryTtl; + + private final ConcurrentHashMap jobs = new ConcurrentHashMap<>(); + + public BuildProgressService(@Value("${extender.progress.enabled:true}") boolean enabled, + @Value("${extender.progress.sse-timeout:1800000}") long sseTimeout, + @Value("${extender.progress.event-buffer-size:256}") int eventBufferSize, + @Value("${extender.progress.max-subscribers-per-job:8}") int maxSubscribersPerJob, + @Value("${extender.progress.registry-ttl:1200000}") long registryTtl) { + this.enabled = enabled; + this.sseTimeout = sseTimeout; + this.eventBufferSize = eventBufferSize; + this.maxSubscribersPerJob = maxSubscribersPerJob; + this.registryTtl = registryTtl; + } + + public boolean isEnabled() { + return enabled; + } + + /** + * Create the progress entry for a job. Must be called before the async + * build is dispatched so that reporterFor() finds it on the worker thread. + */ + public ProgressReporter register(String jobId) { + if (!enabled) { + return ProgressReporter.NOOP; + } + return jobs.computeIfAbsent(jobId, JobProgress::new); + } + + /** Reporter for an already registered job, or NOOP if unknown/disabled. */ + public ProgressReporter reporterFor(String jobId) { + JobProgress job = jobs.get(jobId); + return job != null ? job : ProgressReporter.NOOP; + } + + /** Drop a job that never started building (e.g. failed dispatch). */ + public void remove(String jobId) { + JobProgress job = jobs.remove(jobId); + if (job != null) { + job.completeEmitters(); + } + } + + /** + * Subscribe to a live job. Returns null when the job is unknown. + * Throws IllegalStateException when the job has too many subscribers. + */ + public SseEmitter subscribe(String jobId, Long lastEventId) { + JobProgress job = jobs.get(jobId); + if (job == null) { + return null; + } + return job.subscribe(lastEventId); + } + + /** + * Republish an event relayed from a remote builder under the frontend's + * local jobId. The event is re-sequenced locally. + */ + public void publishRaw(String jobId, BuildStage stage, String detail, int percent, + String extension, Integer currentFile, Integer totalFiles) { + JobProgress job = jobs.get(jobId); + if (job != null) { + job.publish(stage, detail, percent, extension, currentFile, totalFiles); + } + } + + /** + * One-shot emitter for a job that already has its result files on disk: + * sends a single terminal event and completes. + */ + public SseEmitter finishedJobEmitter(String jobId, boolean success) { + SseEmitter emitter = new SseEmitter(sseTimeout); + BuildStage stage = success ? BuildStage.SUCCESS : BuildStage.ERROR; + BuildProgressEvent event = new BuildProgressEvent(jobId, 0, System.currentTimeMillis(), + stage, "Build " + (success ? "succeeded" : "failed"), 100, null, null, null); + try { + emitter.send(SseEmitter.event() + .id(Long.toString(event.getSeq())) + .name(EVENT_NAME) + .data(event, MediaType.APPLICATION_JSON)); + emitter.complete(); + } catch (IOException | IllegalStateException e) { + emitter.completeWithError(e); + } + return emitter; + } + + @Scheduled(fixedDelayString = "${extender.progress.heartbeat-interval:15000}") + public void sendHeartbeats() { + for (JobProgress job : jobs.values()) { + job.heartbeat(); + } + } + + @Scheduled(fixedDelayString = "${extender.progress.cleanup-period:20000}") + public void cleanStaleJobs() { + long deadline = System.currentTimeMillis() - registryTtl; + for (Map.Entry entry : jobs.entrySet()) { + if (entry.getValue().lastTouched < deadline) { + LOGGER.warn("Removing stale progress entry for job {}", entry.getKey()); + remove(entry.getKey()); + } + } + } + + private static String truncate(String detail) { + if (detail != null && detail.length() > MAX_DETAIL_LENGTH) { + return detail.substring(0, MAX_DETAIL_LENGTH); + } + return detail; + } + + // Start/end of the percent budget for each stage. COMPILING is + // interpolated by aggregate file counters; other stages report their + // start value. The global percent is clamped non-decreasing because + // file totals grow while extensions are discovered sequentially. + private static int stageStartPercent(BuildStage stage) { + switch (stage) { + case RECEIVED: + case QUEUED: return 0; + case SDK: return 1; + case DEPENDENCIES: return 10; + case MANIFESTS: return 20; + case PLATFORM: return 25; + case COMPILING: return 35; + case REMOTE_BUILDING: return 35; + case LINKING: return 80; + case PACKAGING: return 92; + case SUCCESS: + case ERROR: return 100; + default: return 0; + } + } + + private static final int COMPILING_START = 35; + private static final int COMPILING_END = 80; + + private static class ExtensionCounter { + final AtomicInteger total = new AtomicInteger(); + final AtomicInteger done = new AtomicInteger(); + } + + private class JobProgress implements ProgressReporter { + private final String jobId; + private final Object lock = new Object(); + private final AtomicLong seq = new AtomicLong(); + private final List emitters = new CopyOnWriteArrayList<>(); + private final Deque buffer = new ArrayDeque<>(); + private final Map extensionCounters = new ConcurrentHashMap<>(); + private final AtomicInteger jobTotalFiles = new AtomicInteger(); + private final AtomicInteger jobCompletedFiles = new AtomicInteger(); + private volatile int lastPercent = 0; + private volatile BuildProgressEvent lastEvent; + private volatile long lastTouched = System.currentTimeMillis(); + + JobProgress(String jobId) { + this.jobId = jobId; + } + + @Override + public void stage(BuildStage stage, String detail) { + publish(stage, detail, stageStartPercent(stage), null, null, null); + } + + @Override + public void compileBatchBegin(String extension, int totalFiles) { + ExtensionCounter counter = extensionCounters.computeIfAbsent(extension, k -> new ExtensionCounter()); + // increment and publish atomically so counters in the event + // stream never go backwards (per-file callbacks are concurrent) + synchronized (lock) { + int extTotal = counter.total.addAndGet(totalFiles); + jobTotalFiles.addAndGet(totalFiles); + publish(BuildStage.COMPILING, extension + ": compiling source files", compilingPercent(), + extension, counter.done.get(), extTotal); + } + } + + @Override + public void fileCompiled(String extension) { + ExtensionCounter counter = extensionCounters.computeIfAbsent(extension, k -> new ExtensionCounter()); + synchronized (lock) { + int extDone = counter.done.incrementAndGet(); + jobCompletedFiles.incrementAndGet(); + publish(BuildStage.COMPILING, extension + ": compiling source files", compilingPercent(), + extension, extDone, counter.total.get()); + } + } + + @Override + public void terminal(boolean success, String detail) { + publish(success ? BuildStage.SUCCESS : BuildStage.ERROR, detail, 100, null, null, null); + completeEmitters(); + jobs.remove(jobId); + } + + private int compilingPercent() { + int total = jobTotalFiles.get(); + if (total <= 0) { + return COMPILING_START; + } + int done = Math.min(jobCompletedFiles.get(), total); + return COMPILING_START + (COMPILING_END - COMPILING_START) * done / total; + } + + void publish(BuildStage stage, String detail, int percent, + String extension, Integer currentFile, Integer totalFiles) { + synchronized (lock) { + // never let the reported percent regress + int clamped = Math.max(lastPercent, percent); + lastPercent = clamped; + BuildProgressEvent event = new BuildProgressEvent(jobId, seq.incrementAndGet(), + System.currentTimeMillis(), stage, truncate(detail), clamped, + extension, currentFile, totalFiles); + lastEvent = event; + lastTouched = event.getTs(); + buffer.addLast(event); + while (buffer.size() > eventBufferSize) { + buffer.removeFirst(); + } + for (SseEmitter emitter : emitters) { + sendEvent(emitter, event); + } + } + } + + SseEmitter subscribe(Long lastEventId) { + SseEmitter emitter = new SseEmitter(sseTimeout); + emitter.onCompletion(() -> emitters.remove(emitter)); + emitter.onTimeout(() -> emitters.remove(emitter)); + emitter.onError(e -> emitters.remove(emitter)); + synchronized (lock) { + if (emitters.size() >= maxSubscribersPerJob) { + throw new IllegalStateException(String.format("Too many progress subscribers for job %s", jobId)); + } + emitters.add(emitter); + // Catch the subscriber up while holding the lock so no live + // event can interleave. If we can replay the exact gap since + // lastEventId, do so; otherwise a snapshot of the current + // state is sufficient (early events precede any subscriber: + // the build is dispatched before the jobId is returned). + List replay = replayableTail(lastEventId); + if (replay != null) { + for (BuildProgressEvent event : replay) { + sendEvent(emitter, event); + } + } else if (lastEvent != null) { + sendEvent(emitter, lastEvent); + } + } + return emitter; + } + + /** + * Buffered events with seq > lastEventId, or null when the gap + * cannot be covered (no id given, or the buffer no longer reaches + * back that far) and a snapshot must be sent instead. + */ + private List replayableTail(Long lastEventId) { + if (lastEventId == null || buffer.isEmpty()) { + return null; + } + if (buffer.peekFirst().getSeq() > lastEventId + 1) { + return null; + } + List tail = new ArrayList<>(); + for (BuildProgressEvent event : buffer) { + if (event.getSeq() > lastEventId) { + tail.add(event); + } + } + return tail; + } + + private void sendEvent(SseEmitter emitter, BuildProgressEvent event) { + try { + emitter.send(SseEmitter.event() + .id(Long.toString(event.getSeq())) + .name(EVENT_NAME) + .data(event, MediaType.APPLICATION_JSON)); + } catch (IOException | IllegalStateException e) { + // dead client; not a server error + emitters.remove(emitter); + emitter.completeWithError(e); + } + } + + void heartbeat() { + synchronized (lock) { + for (SseEmitter emitter : emitters) { + try { + emitter.send(SseEmitter.event().comment("ka")); + } catch (IOException | IllegalStateException e) { + emitters.remove(emitter); + emitter.completeWithError(e); + } + } + } + } + + void completeEmitters() { + synchronized (lock) { + for (SseEmitter emitter : emitters) { + try { + emitter.complete(); + } catch (IllegalStateException e) { + // already completed + } + } + emitters.clear(); + } + } + } +} diff --git a/server/src/main/java/com/defold/extender/progress/BuildStage.java b/server/src/main/java/com/defold/extender/progress/BuildStage.java new file mode 100644 index 00000000..d12a4ede --- /dev/null +++ b/server/src/main/java/com/defold/extender/progress/BuildStage.java @@ -0,0 +1,25 @@ +package com.defold.extender.progress; + +/** + * Stages of an async build, in the order they normally occur. + * REMOTE_BUILDING is a coarse stage used by a frontend instance when the + * remote builder does not expose detailed progress. + */ +public enum BuildStage { + RECEIVED, + QUEUED, + SDK, + DEPENDENCIES, + MANIFESTS, + PLATFORM, + COMPILING, + LINKING, + PACKAGING, + REMOTE_BUILDING, + SUCCESS, + ERROR; + + public boolean isTerminal() { + return this == SUCCESS || this == ERROR; + } +} diff --git a/server/src/main/java/com/defold/extender/progress/ProgressReporter.java b/server/src/main/java/com/defold/extender/progress/ProgressReporter.java new file mode 100644 index 00000000..bf503ff8 --- /dev/null +++ b/server/src/main/java/com/defold/extender/progress/ProgressReporter.java @@ -0,0 +1,30 @@ +package com.defold.extender.progress; + +/** + * Sink for build progress. Implementations must tolerate calls from + * multiple threads: per-file callbacks arrive concurrently from the + * compile thread pool (see ProcessExecutor.executeCommands). + * + * All methods default to no-ops so the build pipeline can be used + * without progress tracking (NOOP is the default in Extender.Builder). + */ +public interface ProgressReporter { + + ProgressReporter NOOP = new ProgressReporter() {}; + + /** The build entered a new stage. */ + default void stage(BuildStage stage, String detail) {} + + /** + * A batch of compile commands is about to run for the given extension + * (or pod). Adds totalFiles to the extension's and the job's file totals. + * May be called more than once per extension (e.g. swift + objc batches). + */ + default void compileBatchBegin(String extension, int totalFiles) {} + + /** One source file of the given extension finished compiling. Thread-safe. */ + default void fileCompiled(String extension) {} + + /** The build finished. Must be reported only after the result/error file is in place. */ + default void terminal(boolean success, String detail) {} +} diff --git a/server/src/main/java/com/defold/extender/remote/RemoteEngineBuilder.java b/server/src/main/java/com/defold/extender/remote/RemoteEngineBuilder.java index 7001a2bf..3ffae205 100644 --- a/server/src/main/java/com/defold/extender/remote/RemoteEngineBuilder.java +++ b/server/src/main/java/com/defold/extender/remote/RemoteEngineBuilder.java @@ -5,6 +5,9 @@ import com.defold.extender.ExtenderException; import com.defold.extender.ExtenderUtil; import com.defold.extender.metrics.MetricsWriter; +import com.defold.extender.progress.BuildProgressService; +import com.defold.extender.progress.BuildStage; +import com.defold.extender.progress.ProgressReporter; import com.defold.extender.services.DataCacheService; import com.defold.extender.services.GCPInstanceService; import com.defold.extender.tracing.ExtenderTracerInterceptor; @@ -56,6 +59,7 @@ public class RemoteEngineBuilder { private static final Logger LOGGER = LoggerFactory.getLogger(RemoteEngineBuilder.class); private GCPInstanceService instanceService; + private BuildProgressService buildProgressService; private File jobResultLocation; private long buildSleepTimeout; private long buildResultWaitTimeout; @@ -63,12 +67,14 @@ public class RemoteEngineBuilder { protected final HttpClient httpClient; public RemoteEngineBuilder(Optional instanceService, + BuildProgressService buildProgressService, @Value("${extender.job-result.location}") String jobResultLocation, @Value("${extender.remote-builder.build-sleep-timeout:5000}") long buildSleepTimeout, @Value("${extender.remote-builder.build-result-wait-timeout:1200000}") long buildResultWaitTimeout, @Autowired Tracer tracer, @Autowired Propagator propogator) { instanceService.ifPresent(val -> { LOGGER.info("Instance client is initialized"); this.instanceService = val; }); + this.buildProgressService = buildProgressService; this.buildSleepTimeout = buildSleepTimeout; this.buildResultWaitTimeout = buildResultWaitTimeout; this.jobResultLocation = new File(jobResultLocation); @@ -108,17 +114,27 @@ public void buildAsync(final RemoteInstanceConfig remoteInstanceConfig, throw new RemoteBuildException("Failed to add files to multipart request", e); } + ProgressReporter progressReporter = buildProgressService.reporterFor(jobName); + RemoteProgressRelay relay = null; try { final String serverUrl = String.format("%s/build_async/%s/%s", remoteInstanceConfig.getUrl(), platform, sdkVersion); final HttpPost request = new HttpPost(serverUrl); request.setEntity(httpEntity); - + touchInstance(remoteInstanceConfig.getInstanceId()); HttpResponse response = httpClient.execute(request); // copied from ExtenderClient. Think about code deduplication. if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) { String jobId = EntityUtils.toString(response.getEntity()); LOGGER.info(String.format("Remote async build posted. Wait job id: %s", jobId)); + if (buildProgressService.isEnabled()) { + // relay the remote builder's progress stream into the local + // registry under the local job id + relay = new RemoteProgressRelay(remoteInstanceConfig.getUrl(), jobId, jobName, buildProgressService); + Thread relayThread = new Thread(relay, String.format("progress-relay-%s", jobName)); + relayThread.setDaemon(true); + relayThread.start(); + } long currentTime = System.currentTimeMillis(); Integer jobStatus = 0; Thread.sleep(buildSleepTimeout); @@ -131,6 +147,12 @@ public void buildAsync(final RemoteInstanceConfig remoteInstanceConfig, LOGGER.info(String.format("Job %s status is %d", jobId, jobStatus)); break; } + if (relay == null || !relay.isDeliveringEvents()) { + // old remote builder (or dropped stream): coarse progress + long elapsedSeconds = (System.currentTimeMillis() - currentTime) / 1000; + buildProgressService.publishRaw(jobName, BuildStage.REMOTE_BUILDING, + String.format("Building remotely, %ds elapsed", elapsedSeconds), 0, null, null, null); + } Thread.sleep(buildSleepTimeout); } if (jobStatus == 0) { @@ -138,6 +160,7 @@ public void buildAsync(final RemoteInstanceConfig remoteInstanceConfig, PrintWriter writer = new PrintWriter(errorFile); writer.write(String.format("Job %s result cannot be defined during %d", jobId, buildResultWaitTimeout)); writer.close(); + progressReporter.terminal(false, "Remote build timed out"); } touchInstance(remoteInstanceConfig.getInstanceId()); HttpGet resultRequest = new HttpGet(String.format("%s/job_result?jobId=%s", remoteInstanceConfig.getUrl(), jobId)); @@ -151,13 +174,16 @@ public void buildAsync(final RemoteInstanceConfig remoteInstanceConfig, os.close(); File targetResult = new File(resultDir, BuilderConstants.BUILD_RESULT_FILENAME); Files.move(tmpResult.toPath(), targetResult.toPath(), StandardCopyOption.ATOMIC_MOVE); - } else { + // terminal only after the local result file is in place + progressReporter.terminal(true, "Build succeeded"); + } else if (jobStatus != 0) { LOGGER.error(Markers.COMPILATION_ERROR, "Failed to build source."); File errorFile = new File(resultDir, BuilderConstants.BUILD_ERROR_FILENAME); PrintWriter writer = new PrintWriter(errorFile); IOUtils.copy(response.getEntity().getContent(), writer, Charset.defaultCharset()); writer.close(); EntityUtils.consumeQuietly(response.getEntity()); + progressReporter.terminal(false, "Build failed"); } } else { LOGGER.error(Markers.COMPILATION_ERROR, "Failed to build source."); @@ -166,6 +192,7 @@ public void buildAsync(final RemoteInstanceConfig remoteInstanceConfig, IOUtils.copy(response.getEntity().getContent(), writer, Charset.defaultCharset()); writer.close(); EntityUtils.consumeQuietly(response.getEntity()); + progressReporter.terminal(false, "Remote build request failed"); } metricsWriter.measureRemoteEngineBuild(buildTimer.start(), platform); } catch (Exception e) { @@ -174,7 +201,11 @@ public void buildAsync(final RemoteInstanceConfig remoteInstanceConfig, writer.write("Failed to communicate with Extender service."); e.printStackTrace(writer); writer.close(); + progressReporter.terminal(false, "Failed to communicate with remote builder"); } finally { + if (relay != null) { + relay.stop(); + } tmpUploadArchive.delete(); metricsWriter.measureRemoteEngineBuild(buildTimer.start(), platform); // Delete temporary upload directory diff --git a/server/src/main/java/com/defold/extender/remote/RemoteProgressRelay.java b/server/src/main/java/com/defold/extender/remote/RemoteProgressRelay.java new file mode 100644 index 00000000..77fe6b98 --- /dev/null +++ b/server/src/main/java/com/defold/extender/remote/RemoteProgressRelay.java @@ -0,0 +1,188 @@ +package com.defold.extender.remote; + +import com.defold.extender.progress.BuildProgressService; +import com.defold.extender.progress.BuildStage; + +import org.apache.http.HttpResponse; +import org.apache.http.HttpStatus; +import org.apache.http.client.config.RequestConfig; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClientBuilder; +import org.apache.http.util.EntityUtils; +import org.json.simple.JSONObject; +import org.json.simple.parser.JSONParser; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; + +/** + * Relays SSE build-progress events from a remote builder into the + * frontend's local progress registry, rewriting the remote jobId to the + * frontend's local jobId (events are re-sequenced by the registry). + * + * Runs on a dedicated daemon thread (never on the build executor: a + * blocked relay must not starve build workers). Uses its own single-use + * HTTP client so the long-lived stream cannot exhaust the shared client's + * connection pool. + * + * The remote terminal event is suppressed: the frontend emits its own + * terminal only after the result file has landed locally, keeping + * /job_status consistent for anyone reacting to the terminal event. + * + * If the remote builder runs an older server version (404) the relay + * gives up silently; RemoteEngineBuilder's poll loop then synthesizes + * coarse REMOTE_BUILDING ticks instead (see isDeliveringEvents()). + */ +public class RemoteProgressRelay implements Runnable { + private static final Logger LOGGER = LoggerFactory.getLogger(RemoteProgressRelay.class); + + private static final int MAX_RECONNECT_ATTEMPTS = 5; + private static final long RECONNECT_BACKOFF_MS = 2000; + + private final String remoteUrl; + private final String remoteJobId; + private final String localJobId; + private final BuildProgressService progressService; + + private volatile boolean stopped = false; + private volatile boolean streaming = false; + private volatile HttpGet currentRequest = null; + private long lastEventId = -1; + + public RemoteProgressRelay(String remoteUrl, String remoteJobId, String localJobId, + BuildProgressService progressService) { + this.remoteUrl = remoteUrl; + this.remoteJobId = remoteJobId; + this.localJobId = localJobId; + this.progressService = progressService; + } + + /** True while events are flowing from the remote builder. */ + public boolean isDeliveringEvents() { + return streaming; + } + + public void stop() { + stopped = true; + HttpGet request = currentRequest; + if (request != null) { + request.abort(); + } + } + + @Override + public void run() { + try (CloseableHttpClient client = HttpClientBuilder.create().build()) { + int attempts = 0; + while (!stopped && attempts < MAX_RECONNECT_ATTEMPTS) { + attempts++; + try { + if (stream(client)) { + return; // unsupported by remote, or terminal seen + } + } catch (IOException e) { + if (stopped) { + return; + } + LOGGER.info("Progress relay for job {} dropped ({}), reconnecting", localJobId, e.getMessage()); + } finally { + streaming = false; + } + Thread.sleep(RECONNECT_BACKOFF_MS); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (Exception e) { + LOGGER.warn("Progress relay for job {} failed", localJobId, e); + } finally { + streaming = false; + } + } + + /** + * Opens the SSE stream and relays events as they arrive. + * Returns true when the relay is done for good (remote has no progress + * support, or a terminal event arrived); false to reconnect. + */ + private boolean stream(CloseableHttpClient client) throws IOException { + HttpGet request = new HttpGet(String.format("%s/job_progress?jobId=%s", remoteUrl, remoteJobId)); + request.setHeader("Accept", "text/event-stream"); + if (lastEventId >= 0) { + request.setHeader("Last-Event-ID", Long.toString(lastEventId)); + } + // the stream is expected to stay open for the whole build; rely on + // the remote's heartbeats and stop() instead of a socket timeout + request.setConfig(RequestConfig.custom().setSocketTimeout(0).build()); + currentRequest = request; + try { + HttpResponse response = client.execute(request); + int status = response.getStatusLine().getStatusCode(); + if (status != HttpStatus.SC_OK) { + EntityUtils.consumeQuietly(response.getEntity()); + if (status == HttpStatus.SC_NOT_FOUND || status == HttpStatus.SC_METHOD_NOT_ALLOWED) { + LOGGER.info("Remote builder at {} does not expose build progress, using coarse fallback", remoteUrl); + return true; + } + LOGGER.info("Progress relay for job {} got status {}", localJobId, status); + return false; // transient; retry + } + streaming = true; + try (BufferedReader reader = new BufferedReader( + new InputStreamReader(response.getEntity().getContent(), StandardCharsets.UTF_8))) { + String eventId = null; + StringBuilder data = new StringBuilder(); + String line; + while (!stopped && (line = reader.readLine()) != null) { + if (line.isEmpty()) { + // end of one SSE event + if (data.length() > 0 && dispatch(eventId, data.toString())) { + return true; // terminal seen, done + } + eventId = null; + data.setLength(0); + } else if (line.startsWith("id:")) { + eventId = line.substring(3).trim(); + } else if (line.startsWith("data:")) { + data.append(line.substring(5).trim()); + } + // "event:" names and ":" comments (heartbeats) are ignored + } + } + return stopped; // EOF mid-build: reconnect + } finally { + currentRequest = null; + } + } + + /** Relays one event. Returns true when it was a terminal event. */ + private boolean dispatch(String eventId, String data) { + try { + JSONObject json = (JSONObject) new JSONParser().parse(data); + if (eventId != null) { + lastEventId = Long.parseLong(eventId); + } + BuildStage stage = BuildStage.valueOf((String) json.get("stage")); + if (stage.isTerminal()) { + // suppressed: the frontend emits its own terminal event + // after the result file lands locally + return true; + } + Number percent = (Number) json.get("percent"); + Number currentFile = (Number) json.get("currentFile"); + Number totalFiles = (Number) json.get("totalFiles"); + progressService.publishRaw(localJobId, stage, (String) json.get("detail"), + percent != null ? percent.intValue() : 0, + (String) json.get("extension"), + currentFile != null ? currentFile.intValue() : null, + totalFiles != null ? totalFiles.intValue() : null); + } catch (Exception e) { + LOGGER.debug("Ignoring malformed progress event for job {}: {}", localJobId, e.getMessage()); + } + return false; + } +} diff --git a/server/src/main/resources/application.yml b/server/src/main/resources/application.yml index 08b9fe57..22eda527 100644 --- a/server/src/main/resources/application.yml +++ b/server/src/main/resources/application.yml @@ -50,6 +50,19 @@ extender: location: /tmp/results cleanup-period: 20000 lifetime: 1200000 + # live build progress over Server-Sent Events (GET /job_progress?jobId=) + progress: + enabled: true + # SseEmitter timeout, ms (~ remote-builder build-result-wait-timeout) + sse-timeout: 1800000 + # SSE comment keepalive, ms (must be < jetty connection-idle-timeout) + heartbeat-interval: 15000 + # per-job replay ring buffer for Last-Event-ID reconnects + event-buffer-size: 256 + max-subscribers-per-job: 8 + # orphan sweep for jobs that died without a terminal event + registry-ttl: 1200000 + cleanup-period: 20000 # see ExtenderController.InstanceType enum # FRONTEND_ONLY, BUILDER_ONLY, MIXED instance-type: MIXED diff --git a/server/src/test/java/com/defold/extender/BuildProgressControllerTest.java b/server/src/test/java/com/defold/extender/BuildProgressControllerTest.java new file mode 100644 index 00000000..34d02917 --- /dev/null +++ b/server/src/test/java/com/defold/extender/BuildProgressControllerTest.java @@ -0,0 +1,209 @@ +package com.defold.extender; + +import com.defold.extender.progress.BuildProgressService; +import com.defold.extender.progress.BuildStage; +import com.defold.extender.progress.ProgressReporter; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.MvcResult; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; + +import java.io.File; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.request; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +public class BuildProgressControllerTest { + + @TempDir + File jobResultLocation; + + private BuildProgressService service; + private MockMvc mockMvc; + + @BeforeEach + public void setUp() { + service = new BuildProgressService(true, 30000, 256, 8, 1200000); + mockMvc = mockMvcFor(service); + } + + private MockMvc mockMvcFor(BuildProgressService service) { + BuildProgressController controller = new BuildProgressController(service, jobResultLocation.getAbsolutePath()); + return MockMvcBuilders.standaloneSetup(controller).build(); + } + + private MvcResult subscribe(String jobId) throws Exception { + return mockMvc.perform(get("/job_progress").param("jobId", jobId)) + .andExpect(request().asyncStarted()) + .andReturn(); + } + + private static List eventIds(String sseContent) { + List ids = new ArrayList<>(); + Matcher matcher = Pattern.compile("^id:(\\d+)$", Pattern.MULTILINE).matcher(sseContent); + while (matcher.find()) { + ids.add(Long.parseLong(matcher.group(1))); + } + return ids; + } + + @Test + public void testSnapshotAndLiveEvents() throws Exception { + ProgressReporter reporter = service.register("job1"); + reporter.stage(BuildStage.SDK, "Downloading Defold SDK"); + + MvcResult result = subscribe("job1"); + String content = result.getResponse().getContentAsString(); + // snapshot of the current state arrives immediately + assertTrue(content.contains("\"stage\":\"SDK\"")); + + reporter.stage(BuildStage.COMPILING, "extension1"); + reporter.terminal(true, "Build succeeded"); + + content = result.getResponse().getContentAsString(); + assertTrue(content.contains("\"stage\":\"COMPILING\"")); + assertTrue(content.contains("\"stage\":\"SUCCESS\"")); + assertTrue(content.contains("\"terminal\":true")); + assertTrue(content.contains("\"percent\":100")); + + // seq strictly increasing + List ids = eventIds(content); + assertFalse(ids.isEmpty()); + for (int i = 1; i < ids.size(); i++) { + assertTrue(ids.get(i) > ids.get(i - 1), "event ids must be strictly increasing"); + } + } + + @Test + public void testLastEventIdReplay() throws Exception { + ProgressReporter reporter = service.register("job1"); + reporter.stage(BuildStage.SDK, "one"); // seq 1 + reporter.stage(BuildStage.DEPENDENCIES, "two"); // seq 2 + reporter.stage(BuildStage.MANIFESTS, "three"); // seq 3 + reporter.stage(BuildStage.PLATFORM, "four"); // seq 4 + + MvcResult result = mockMvc.perform(get("/job_progress").param("jobId", "job1") + .header("Last-Event-ID", "2")) + .andExpect(request().asyncStarted()) + .andReturn(); + + assertEquals(List.of(3L, 4L), eventIds(result.getResponse().getContentAsString())); + } + + @Test + public void testSnapshotWhenReplayGapExceedsBuffer() throws Exception { + // buffer of 2 only holds seq 4 and 5; Last-Event-ID 1 cannot be replayed + service = new BuildProgressService(true, 30000, 2, 8, 1200000); + mockMvc = mockMvcFor(service); + ProgressReporter reporter = service.register("job1"); + for (int i = 0; i < 5; i++) { + reporter.stage(BuildStage.SDK, "event " + i); // seq 1..5 + } + + MvcResult result = mockMvc.perform(get("/job_progress").param("jobId", "job1") + .header("Last-Event-ID", "1")) + .andExpect(request().asyncStarted()) + .andReturn(); + + // only the state snapshot (the latest event) + assertEquals(List.of(5L), eventIds(result.getResponse().getContentAsString())); + } + + @Test + public void testConcurrentFileCountingAndPercentClamp() throws Exception { + ProgressReporter reporter = service.register("job1"); + reporter.stage(BuildStage.COMPILING, "extension1"); + reporter.compileBatchBegin("extension1", 100); + + ExecutorService executor = Executors.newFixedThreadPool(8); + for (int i = 0; i < 100; i++) { + executor.submit(() -> reporter.fileCompiled("extension1")); + } + executor.shutdown(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + + // all 100 files counted exactly once: compiling budget is exhausted + MvcResult result = subscribe("job1"); + String content = result.getResponse().getContentAsString(); + assertTrue(content.contains("\"currentFile\":100")); + assertTrue(content.contains("\"totalFiles\":100")); + assertTrue(content.contains("\"percent\":80")); + + // a late-discovered extension grows the totals; percent must not regress + reporter.compileBatchBegin("extension2", 100); + content = result.getResponse().getContentAsString(); + String lastData = content.substring(content.lastIndexOf("data:")); + assertTrue(lastData.contains("\"extension\":\"extension2\"")); + assertTrue(lastData.contains("\"percent\":80"), "percent must stay clamped at 80, got: " + lastData); + } + + @Test + public void testFinishedJobFromResultFiles() throws Exception { + File jobDir = new File(jobResultLocation, "job42"); + assertTrue(jobDir.mkdir()); + Files.createFile(new File(jobDir, BuilderConstants.BUILD_RESULT_FILENAME).toPath()); + + MvcResult result = subscribe("job42"); + String content = result.getResponse().getContentAsString(); + assertTrue(content.contains("\"stage\":\"SUCCESS\"")); + assertTrue(content.contains("\"terminal\":true")); + } + + @Test + public void testFinishedJobWithErrorFile() throws Exception { + File jobDir = new File(jobResultLocation, "job43"); + assertTrue(jobDir.mkdir()); + Files.createFile(new File(jobDir, BuilderConstants.BUILD_ERROR_FILENAME).toPath()); + + MvcResult result = subscribe("job43"); + String content = result.getResponse().getContentAsString(); + assertTrue(content.contains("\"stage\":\"ERROR\"")); + assertTrue(content.contains("\"terminal\":true")); + } + + @Test + public void testUnknownJobReturns404() throws Exception { + mockMvc.perform(get("/job_progress").param("jobId", "nosuchjob")) + .andExpect(status().isNotFound()); + } + + @Test + public void testMalformedJobIdReturns404() throws Exception { + mockMvc.perform(get("/job_progress").param("jobId", "../../etc/passwd")) + .andExpect(status().isNotFound()); + } + + @Test + public void testDisabledReturns404() throws Exception { + service = new BuildProgressService(false, 30000, 256, 8, 1200000); + mockMvc = mockMvcFor(service); + service.register("job1"); // no-op when disabled + mockMvc.perform(get("/job_progress").param("jobId", "job1")) + .andExpect(status().isNotFound()); + } + + @Test + public void testTooManySubscribersReturns429() throws Exception { + service = new BuildProgressService(true, 30000, 256, 1, 1200000); + mockMvc = mockMvcFor(service); + service.register("job1"); + subscribe("job1"); + mockMvc.perform(get("/job_progress").param("jobId", "job1")) + .andExpect(status().isTooManyRequests()); + } +} diff --git a/server/src/test/java/com/defold/extender/IntegrationTest.java b/server/src/test/java/com/defold/extender/IntegrationTest.java index c2b0c217..c3262755 100644 --- a/server/src/test/java/com/defold/extender/IntegrationTest.java +++ b/server/src/test/java/com/defold/extender/IntegrationTest.java @@ -34,6 +34,8 @@ import java.nio.file.Path; import java.nio.file.StandardCopyOption; import java.util.*; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.zip.ZipEntry; import java.util.zip.ZipFile; @@ -295,6 +297,98 @@ public void buildEngine(TestConfiguration configuration) throws IOException, Ext doBuild(sourceFiles, configuration); } + private static class ProgressEvent { + final String stage; + final int percent; + final int currentFile; + final int totalFiles; + + ProgressEvent(String stage, int percent, int currentFile, int totalFiles) { + this.stage = stage; + this.percent = percent; + this.currentFile = currentFile; + this.totalFiles = totalFiles; + } + } + + @ParameterizedTest(name = "[{index}] {displayName} {arguments}") + @MethodSource("data") + public void buildEngineWithProgress(TestConfiguration configuration) throws IOException, ExtenderClientException, InterruptedException { + List sourceFiles = Lists.newArrayList( + new FileExtenderResource("test-data/AndroidManifest.xml", "AndroidManifest.xml"), + new FileExtenderResource("test-data/ext2/ext.manifest"), + new FileExtenderResource("test-data/ext2/src/test_ext.cpp"), + new FileExtenderResource(String.format("test-data/ext2/lib/%s/%s", configuration.platform, getLibName(configuration.platform, "alib"))), + new FileExtenderResource(String.format("test-data/ext2/lib/%s/%s", configuration.platform, getLibName(configuration.platform, "blib"))) + ); + + File cacheDir = Files.createTempDirectory(String.format("progress-%s-%s", configuration.platform, configuration.version.toString())).toFile(); + cacheDir.deleteOnExit(); + ExtenderClient extenderClient = new ExtenderClient("http://localhost:" + EXTENDER_PORT, cacheDir); + File destination = Files.createTempFile("dmengine", ".zip").toFile(); + File log = Files.createTempFile("dmengine", ".log").toFile(); + + List events = Collections.synchronizedList(new ArrayList<>()); + CountDownLatch terminalSeen = new CountDownLatch(1); + ExtenderProgressListener listener = (stage, detail, percent, currentFile, totalFiles) -> { + events.add(new ProgressEvent(stage, percent, currentFile, totalFiles)); + if ("SUCCESS".equals(stage) || "ERROR".equals(stage)) { + terminalSeen.countDown(); + } + }; + + try { + extenderClient.build( + configuration.platform, + configuration.version.sha1, + sourceFiles, + destination, + log, + listener + ); + } catch (ExtenderClientException e) { + System.out.println("ERROR LOG:"); + System.out.println(new String(Files.readAllBytes(log.toPath()))); + throw e; + } + + assertTrue(destination.length() > 0, "Resulting engine should be of a size greater than zero."); + + // the terminal event is pushed before /job_status flips, so it should + // already be here (or arrive momentarily) + assertTrue(terminalSeen.await(10, TimeUnit.SECONDS), "Progress listener never saw a terminal event"); + List snapshot = new ArrayList<>(events); + assertTrue(snapshot.size() >= 2, "Expected multiple progress events, got " + snapshot.size()); + assertEquals("SUCCESS", snapshot.get(snapshot.size() - 1).stage); + assertEquals(100, snapshot.get(snapshot.size() - 1).percent); + + // percent must never decrease + int lastPercent = 0; + for (ProgressEvent event : snapshot) { + assertTrue(event.percent >= lastPercent, + String.format("Percent regressed from %d to %d at stage %s", lastPercent, event.percent, event.stage)); + lastPercent = event.percent; + } + + // per-file compile progress: some extension reported file counts and finished them + boolean sawFileCounts = snapshot.stream().anyMatch(e -> "COMPILING".equals(e.stage) && e.totalFiles > 0); + boolean sawCompleted = snapshot.stream().anyMatch(e -> "COMPILING".equals(e.stage) && e.totalFiles > 0 && e.currentFile == e.totalFiles); + assertTrue(sawFileCounts, "Expected COMPILING events with file counts"); + assertTrue(sawCompleted, "Expected a COMPILING event with all files completed"); + + // stages appear in pipeline order for the ones we saw + List stageOrder = Arrays.asList("RECEIVED", "QUEUED", "SDK", "DEPENDENCIES", "MANIFESTS", "PLATFORM", "COMPILING", "LINKING", "PACKAGING"); + int lastIndex = -1; + for (ProgressEvent event : snapshot) { + int index = stageOrder.indexOf(event.stage); + if (index >= 0 && !"COMPILING".equals(event.stage)) { + assertTrue(index >= lastIndex, + String.format("Stage %s arrived after a later stage", event.stage)); + lastIndex = Math.max(lastIndex, index); + } + } + } + @ParameterizedTest(name = "[{index}] {displayName} {arguments}") @MethodSource("data") public void buildExtensionStdLib(TestConfiguration configuration) throws IOException, ExtenderClientException { diff --git a/server/src/test/java/com/defold/extender/process/ProcessExecutorTest.java b/server/src/test/java/com/defold/extender/process/ProcessExecutorTest.java new file mode 100644 index 00000000..ee3329f1 --- /dev/null +++ b/server/src/test/java/com/defold/extender/process/ProcessExecutorTest.java @@ -0,0 +1,43 @@ +package com.defold.extender.process; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class ProcessExecutorTest { + + @Test + public void testExecuteCommandsCallbackFiresOncePerCommand() throws Exception { + ProcessExecutor processExecutor = new ProcessExecutor(); + List commands = new ArrayList<>(); + for (int i = 0; i < 16; i++) { + commands.add("echo file" + i); + } + AtomicInteger completed = new AtomicInteger(); + ProcessExecutor.executeCommands(processExecutor, commands, completed::incrementAndGet); + assertEquals(16, completed.get()); + } + + @Test + public void testExecuteCommandsWithoutCallback() throws Exception { + ProcessExecutor processExecutor = new ProcessExecutor(); + List commands = List.of("echo a", "echo b"); + // the 2-arg overload must still work unchanged + ProcessExecutor.executeCommands(processExecutor, commands); + } + + @Test + public void testFailingCommandStillThrows() { + ProcessExecutor processExecutor = new ProcessExecutor(); + List commands = List.of("echo ok", "false"); + AtomicInteger completed = new AtomicInteger(); + assertThrows(IOException.class, + () -> ProcessExecutor.executeCommands(processExecutor, commands, completed::incrementAndGet)); + } +} diff --git a/server/src/test/java/com/defold/extender/progress/BuildProgressServiceTest.java b/server/src/test/java/com/defold/extender/progress/BuildProgressServiceTest.java new file mode 100644 index 00000000..b7e554fa --- /dev/null +++ b/server/src/test/java/com/defold/extender/progress/BuildProgressServiceTest.java @@ -0,0 +1,90 @@ +package com.defold.extender.progress; + +import org.junit.jupiter.api.Test; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class BuildProgressServiceTest { + + private static BuildProgressService createService() { + return new BuildProgressService(true, 30000, 256, 8, 1200000); + } + + @Test + public void testDisabledServiceReturnsNoop() { + BuildProgressService service = new BuildProgressService(false, 30000, 256, 8, 1200000); + assertSame(ProgressReporter.NOOP, service.register("job1")); + assertSame(ProgressReporter.NOOP, service.reporterFor("job1")); + assertNull(service.subscribe("job1", null)); + } + + @Test + public void testRegisterReturnsSameReporter() { + BuildProgressService service = createService(); + ProgressReporter reporter = service.register("job1"); + assertNotSame(ProgressReporter.NOOP, reporter); + assertSame(reporter, service.register("job1")); + assertSame(reporter, service.reporterFor("job1")); + } + + @Test + public void testUnknownJobReturnsNoopAndNullSubscription() { + BuildProgressService service = createService(); + assertSame(ProgressReporter.NOOP, service.reporterFor("unknown")); + assertNull(service.subscribe("unknown", null)); + } + + @Test + public void testRemoveDropsJob() { + BuildProgressService service = createService(); + service.register("job1"); + assertNotNull(service.subscribe("job1", null)); + service.remove("job1"); + assertNull(service.subscribe("job1", null)); + } + + @Test + public void testTerminalRemovesJob() { + BuildProgressService service = createService(); + ProgressReporter reporter = service.register("job1"); + reporter.stage(BuildStage.SDK, "sdk"); + reporter.terminal(true, "done"); + assertNull(service.subscribe("job1", null)); + // late reports after terminal must be harmless no-ops + service.publishRaw("job1", BuildStage.COMPILING, "late", 50, null, null, null); + } + + @Test + public void testSubscriberLimit() { + BuildProgressService service = new BuildProgressService(true, 30000, 256, 2, 1200000); + service.register("job1"); + assertNotNull(service.subscribe("job1", null)); + assertNotNull(service.subscribe("job1", null)); + assertThrows(IllegalStateException.class, () -> service.subscribe("job1", null)); + } + + @Test + public void testStaleJobSweep() throws InterruptedException { + // ttl 0: everything is stale as soon as it is a millisecond old + BuildProgressService service = new BuildProgressService(true, 30000, 256, 8, 0); + service.register("job1"); + Thread.sleep(5); + service.cleanStaleJobs(); + assertNull(service.subscribe("job1", null)); + } + + @Test + public void testHeartbeatWithSubscribers() { + BuildProgressService service = createService(); + service.register("job1"); + SseEmitter emitter = service.subscribe("job1", null); + assertNotNull(emitter); + // must not throw with live (uninitialized) emitters + service.sendHeartbeats(); + } +} diff --git a/server/src/test/java/com/defold/extender/remote/RemoteProgressRelayTest.java b/server/src/test/java/com/defold/extender/remote/RemoteProgressRelayTest.java new file mode 100644 index 00000000..7ecfbf04 --- /dev/null +++ b/server/src/test/java/com/defold/extender/remote/RemoteProgressRelayTest.java @@ -0,0 +1,86 @@ +package com.defold.extender.remote; + +import com.defold.extender.progress.BuildProgressService; +import com.defold.extender.progress.BuildStage; + +import com.github.tomakehurst.wiremock.WireMockServer; +import com.github.tomakehurst.wiremock.core.WireMockConfiguration; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import static com.github.tomakehurst.wiremock.client.WireMock.aResponse; +import static com.github.tomakehurst.wiremock.client.WireMock.equalTo; +import static com.github.tomakehurst.wiremock.client.WireMock.get; +import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +public class RemoteProgressRelayTest { + + private WireMockServer wireMock; + private BuildProgressService progressService; + + @BeforeEach + public void setUp() { + wireMock = new WireMockServer(WireMockConfiguration.options().dynamicPort()); + wireMock.start(); + progressService = Mockito.mock(BuildProgressService.class); + } + + @AfterEach + public void tearDown() { + wireMock.stop(); + } + + private static String sseEvent(long seq, String json) { + return String.format("id:%d\ndata:%s\n\n", seq, json); + } + + @Test + public void testRelaysEventsWithRewrittenJobIdAndSuppressesTerminal() { + String body = ":ka\n\n" + + sseEvent(1, "{\"jobId\":\"remoteJob\",\"seq\":1,\"stage\":\"SDK\",\"detail\":\"Downloading\",\"percent\":1,\"terminal\":false}") + + sseEvent(2, "{\"jobId\":\"remoteJob\",\"seq\":2,\"stage\":\"COMPILING\",\"detail\":\"ext1\",\"percent\":40,\"extension\":\"ext1\",\"currentFile\":2,\"totalFiles\":10,\"terminal\":false}") + + sseEvent(3, "{\"jobId\":\"remoteJob\",\"seq\":3,\"stage\":\"SUCCESS\",\"detail\":\"done\",\"percent\":100,\"terminal\":true}"); + wireMock.stubFor(get(urlPathEqualTo("/job_progress")) + .withQueryParam("jobId", equalTo("remoteJob")) + .willReturn(aResponse() + .withStatus(200) + .withHeader("Content-Type", "text/event-stream") + .withBody(body))); + + RemoteProgressRelay relay = new RemoteProgressRelay(wireMock.baseUrl(), "remoteJob", "localJob", progressService); + relay.run(); + + verify(progressService).publishRaw(eq("localJob"), eq(BuildStage.SDK), eq("Downloading"), eq(1), + eq(null), eq(null), eq(null)); + verify(progressService).publishRaw(eq("localJob"), eq(BuildStage.COMPILING), eq("ext1"), eq(40), + eq("ext1"), eq(2), eq(10)); + // the remote terminal event is suppressed: the frontend emits its own + verify(progressService, never()).publishRaw(anyString(), eq(BuildStage.SUCCESS), anyString(), anyInt(), any(), any(), any()); + verify(progressService, never()).publishRaw(anyString(), eq(BuildStage.ERROR), anyString(), anyInt(), any(), any(), any()); + } + + @Test + public void testOldRemoteBuilderWithout404GivesUpSilently() { + wireMock.stubFor(get(urlPathEqualTo("/job_progress")) + .willReturn(aResponse().withStatus(404))); + + RemoteProgressRelay relay = new RemoteProgressRelay(wireMock.baseUrl(), "remoteJob", "localJob", progressService); + long start = System.currentTimeMillis(); + relay.run(); + + // gives up immediately (no reconnect loop) and relays nothing + assertFalse(System.currentTimeMillis() - start > 5000, "relay must not retry on 404"); + verify(progressService, never()).publishRaw(anyString(), any(), anyString(), anyInt(), any(), any(), any()); + assertFalse(relay.isDeliveringEvents()); + } +} From 432e4827ab7069ad9d7973f36dd2547bb738f50c Mon Sep 17 00:00:00 2001 From: Kharkunov Eugene Date: Wed, 12 Aug 2026 12:30:00 +0300 Subject: [PATCH 2/2] Guard progress-reconnect-attempts property against malformed values A non-numeric -Dcom.defold.extender.client.progress-reconnect-attempts value made the ExtenderProgressConsumer constructor throw NumberFormatException (CodeQL alert 328). Parse defensively and fall back to the default of 5, logging a warning. --- .../client/ExtenderProgressConsumer.java | 23 +++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/client/src/main/java/com/defold/extender/client/ExtenderProgressConsumer.java b/client/src/main/java/com/defold/extender/client/ExtenderProgressConsumer.java index e60a31d8..acd6f939 100644 --- a/client/src/main/java/com/defold/extender/client/ExtenderProgressConsumer.java +++ b/client/src/main/java/com/defold/extender/client/ExtenderProgressConsumer.java @@ -34,6 +34,8 @@ interface GetRequestFactory { } private static final long RECONNECT_BACKOFF_MS = 2000; + private static final String RECONNECT_ATTEMPTS_PROPERTY = "com.defold.extender.client.progress-reconnect-attempts"; + private static final int DEFAULT_RECONNECT_ATTEMPTS = 5; private final HttpClient httpClient; private final String jobProgressUrl; @@ -51,8 +53,25 @@ interface GetRequestFactory { this.jobProgressUrl = String.format("%s/job_progress?jobId=%s", extenderBaseUrl, jobId); this.requestFactory = requestFactory; this.listener = listener; - this.maxReconnectAttempts = Integer.parseInt( - System.getProperty("com.defold.extender.client.progress-reconnect-attempts", "5")); + this.maxReconnectAttempts = resolveMaxReconnectAttempts(); + } + + /** + * Reads the reconnect-attempts override from a system property, falling back + * to the default when the property is absent or not a valid integer. + */ + private static int resolveMaxReconnectAttempts() { + String value = System.getProperty(RECONNECT_ATTEMPTS_PROPERTY); + if (value == null) { + return DEFAULT_RECONNECT_ATTEMPTS; + } + try { + return Integer.parseInt(value.trim()); + } catch (NumberFormatException e) { + logger.log(Level.WARNING, "Ignoring malformed {0} value ''{1}''; using default {2}", + new Object[] { RECONNECT_ATTEMPTS_PROPERTY, value, DEFAULT_RECONNECT_ATTEMPTS }); + return DEFAULT_RECONNECT_ATTEMPTS; + } } /** Stops the consumer and unblocks the stream read. Safe to call more than once. */