Resume node downloads and lengthen the enrollment token

apps/web/e2e/fixture-console.mjs

@@ -436,14 +436,14 @@ async function sandboxRoute(request, response, path, url) {
   if (path === "/nodes") return send(response, 200, { data: state.nodes });
   if (path === "/enrollment-tokens" && request.method === "POST") {
     if (state.deployment.reset) return error(response, 409, "A sandbox reset is in progress.", "sandbox_reset_in_progress");
-    // As Core: a one-time token valid for ten minutes, whose limits and enrollment ID the node it enrolls takes;
+    // As Core: a one-time token valid for one hour, whose limits and enrollment ID the node it enrolls takes;
     // only microsandbox keeps a retained limit above the active one.
     const input = await body(request);
     const maxActive = input.max_active ?? 1;
     const serial = state.nextId++;
     const enrollmentId = `00000000-0000-4000-8000-${String(serial).padStart(12, "0")}`;
     state.enrollment = { max_active: maxActive, max_retained: state.deployment?.provider === "microsandbox" ? input.max_retained ?? maxActive : maxActive, enrollment_id: enrollmentId };
-    return send(response, 201, { token: `enroll_fixture_${serial}`, expires_at: new Date(Date.now() + 10 * 60_000).toISOString(), enrollment_id: enrollmentId });
+    return send(response, 201, { token: `enroll_fixture_${serial}`, expires_at: new Date(Date.now() + 60 * 60_000).toISOString(), enrollment_id: enrollmentId });
   }
   let m;
   if ((m = path.match(/^\/nodes\/([^/]+)\/allocations$/))) return send(response, 200, { data: state.allocations.filter((entry) => entry.node_id === m[1]) });

apps/web/e2e/nodes.spec.ts

@@ -6,11 +6,11 @@ test.afterEach(async ({ request }) => expectManagementBoundary(request));
 
 test("adds a node: host requirements, a root/sudo command, a countdown, the same command after closing, a new one after expiry, then its own node's registration", async ({ page, request }) => {
   await page.clock.install();
-  // Core counts a token's ten minutes on its own clock; the page's clock stands in for it, so fast-forwarding expires a command.
+  // Core counts a token's hour on its own clock; the page's clock stands in for it, so fast-forwarding expires a command.
   await page.route("**/core/v1/sandbox/enrollment-tokens", async (route) => {
     const response = await route.fetch();
     const now = await page.evaluate(() => Date.now());
-    await route.fulfill({ response, json: { ...await response.json(), expires_at: new Date(now + 10 * 60_000).toISOString() } });
+    await route.fulfill({ response, json: { ...await response.json(), expires_at: new Date(now + 60 * 60_000).toISOString() } });
   });
   await openConsole(page, request, "nodes");
   await page.getByRole("button", { name: "Add node" }).click();
@@ -39,7 +39,7 @@ test("adds a node: host requirements, a root/sudo command, a countdown, the same
   await expect(field).toHaveValue(/curl [^\n]* 'https:\/\/core\.example\.com\/node-install\/node-install\.pyz' /);
   await expect(field).toHaveValue(/ --source-url 'https:\/\/core\.example\.com' --core-url 'https:\/\/core\.example\.com' /);
   await expect(add.getByText("If the command is interrupted or the download stalls, run it again: unfinished downloads restart, and verified files are reused.")).toBeVisible();
-  await expect(add.getByRole("timer")).toHaveText(/^Expires in (10:00|9:\d\d)$/);
+  await expect(add.getByRole("timer")).toHaveText(/^Expires in (1:00:00|59:\d\d)$/);
   const progress = add.getByRole("status", { name: "Registration progress" });
   await expect(progress).toHaveText(/Waiting for registration.*Connect.*Docker check/);
 
@@ -55,7 +55,7 @@ test("adds a node: host requirements, a root/sudo command, a countdown, the same
   await expect(progress).toHaveText(/Waiting for registration.*Connect.*Docker check/);
 
   // Once expired, a new command is issued only on request, for the same limits.
-  await page.clock.fastForward("10:30");
+  await page.clock.fastForward("01:00:30");
   await expect(progress).toContainText("Command expired");
   await expect(field).toBeHidden();
   const reissued = tokenRequest();
@@ -69,7 +69,7 @@ test("adds a node: host requirements, a root/sudo command, a countdown, the same
   await add.getByRole("button", { name: "Close dialog" }).click();
   await expect(add).toBeHidden();
   await setNode(request, { id: "node-new", name: "edge-04" });
-  await page.clock.fastForward("10:30");
+  await page.clock.fastForward("01:00:30");
   // Every render from the reopening on is watched, so even a brief "Command expired" would count.
   await page.evaluate(() => {
     const watch = window as unknown as { expiredShown: boolean; expiredWatch: MutationObserver };

deploy/node/README.md

@@ -40,7 +40,7 @@ The distribution manifest is the one download contract for nodes: flat versioned
 - A Compose installation keeps only the node metadata from its release archive; Core's image never acquires execution-only payloads.
 - A node obtains bootstrap metadata from the console that generated its command. Web serves artifacts it has locally and redirects missing declared execution artifacts to the versioned HTTPS release base in the verified manifest. Web never downloads or caches those bytes.
 - Only artifact requests may follow HTTPS redirects, and only without credentials or cookies. Metadata and enrollment requests stay on the configured console. The console publishes only fixed non-secret files and declared artifact names.
-- Download into private temporary files, verify size and SHA-256 before an atomic rename, clear leftover temporary downloads and extracted files before retrying, and reuse only verified cache entries or exact image identities. Never select a release other than the pinned one.
+- Download into private temporary files, verify size and SHA-256 before an atomic rename, resume a leftover partial artifact download with an HTTP range (a server that cannot resume, or bytes that fail verification, restart the file) and clear leftover extracted files before retrying, and reuse only verified cache entries or exact image identities. Never select a release other than the pinned one.
 - Release downloads are anonymous. Never add repository credentials to installed node or Runtime configuration.
 - Manual builds use the `build-<full SHA>` release tag and tag builds the `v*` tag. The manifest's download base must match the release tag; artifact file names and source provenance keep the full source SHA.
 

deploy/node/distribution.py

@@ -114,8 +114,9 @@ def redirect_request(self, request, fp, code, msg, headers, newurl):
         safe_url(newurl)
         if urlsplit(newurl).scheme != 'https' or request.get_method() not in ('GET', 'HEAD'):
             raise ArtifactError('Artifact redirects require HTTPS')
-        # Artifact redirects carry no credentials, cookies or custom headers.
-        return urllib.request.Request(newurl, method=request.get_method(), unverifiable=True)
+        # Artifact redirects carry no credentials, cookies or custom headers; only a resume range follows.
+        headers = {'Range': request.get_header('Range')} if request.has_header('Range') else {}
+        return urllib.request.Request(newurl, headers=headers, method=request.get_method(), unverifiable=True)
 
 
 class NoRedirect(urllib.request.HTTPRedirectHandler):
@@ -156,29 +157,57 @@ def obtain_artifact(manifest, logical_path, destination, replace=False):
     if not isinstance(base, str) or not base or urlsplit(base).query:
         raise ArtifactError('No downloadable artifact source: ' + logical_path)
     url = safe_url(base.rstrip('/') + '/' + entry['filename'])
-    with temporary_file(target) as partial:
-        for attempt in range(3):
-            try:
-                download_artifact(url, partial, entry, logical_path)
-                break
-            except urllib.error.HTTPError as error:
-                if error.code not in (408, 429, 500, 502, 503, 504) or attempt == 2:
-                    raise ArtifactError(f'Artifact download failed (HTTP {error.code}): {logical_path}. Check the console and retry.') from None
-            except (urllib.error.URLError, socket.timeout, ConnectionError, http.client.HTTPException):
-                if attempt == 2:
-                    raise ArtifactError('Artifact transfer interrupted: ' + logical_path + '. Check network access and rerun.') from None
-            time.sleep(attempt + 1)
-        if digest(partial) != entry['sha256']:
+    partial = partial_path(target)
+    failures = 0
+    while True:
+        had = resumable(partial, entry['size'])
+        try:
+            if had < entry['size']:
+                download_artifact(url, partial, entry, logical_path, had)
+        except urllib.error.HTTPError as error:
+            failures += 1
+            if error.code not in (408, 429, 500, 502, 503, 504) or failures == 3:
+                raise ArtifactError(f'Artifact download failed (HTTP {error.code}): {logical_path}. Check the console and retry.') from None
+            time.sleep(failures)
+            continue
+        except (urllib.error.URLError, socket.timeout, ConnectionError, http.client.HTTPException):
+            failures += 1
+            if failures == 3:
+                raise ArtifactError('Artifact transfer interrupted: ' + logical_path + '. Check network access and rerun; the download continues where it stopped.') from None
+            time.sleep(failures)
+            continue
+        if digest(partial) == entry['sha256']:
+            break
+        partial.unlink()
+        if not had:
             raise ArtifactError('Artifact checksum mismatch: ' + logical_path)
-        os.chmod(partial, 0o700 if logical_path.startswith('native/') else 0o600)
-        os.replace(partial, target)
-        return target
+        print(f'The earlier partial download of {logical_path} does not match this release; downloading it again.', flush=True)
+    os.chmod(partial, 0o700 if logical_path.startswith('native/') else 0o600)
+    os.replace(partial, target)
+    return target
+
+
+def partial_path(target):
+    return target.with_name('.' + target.name + '.partial')
+
+
+def resumable(partial, size):
+    """The bytes an earlier interrupted download of this file left, 0 when none is usable."""
+    if partial.is_symlink() or (partial.exists() and
+            (not partial.is_file() or partial.stat().st_uid != os.getuid())):
+        raise DistributionError('Temporary file must be an owned regular file: ' + str(partial))
+    if not partial.exists():
+        return 0
+    if partial.stat().st_size > size:
+        partial.unlink()
+        return 0
+    return partial.stat().st_size
 
 
 @contextmanager
 def temporary_file(target):
     """Clear unfinished work under the caller's installation lock."""
-    partial = target.with_name('.' + target.name + '.partial')
+    partial = partial_path(target)
     if partial.is_symlink() or (partial.exists() and
             (not partial.is_file() or partial.stat().st_uid != os.getuid())):
         raise DistributionError('Temporary file must be an owned regular file: ' + str(target))
@@ -193,33 +222,51 @@ def temporary_file(target):
 SLOW_SECONDS, SLOW_BYTES = 60, 64 * 1024
 
 
-def download_artifact(url, partial, entry, logical_path):
-    """Download a complete artifact into a disposable file."""
+def download_artifact(url, partial, entry, logical_path, have=0):
+    """Download an artifact into its partial file, continuing after the first `have` bytes when the server allows it."""
     size = entry['size']
     opener = urllib.request.build_opener(ArtifactRedirect())
-    with opener.open(url, timeout=60) as stream:
-        flags = os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW | os.O_TRUNC
+    mib = 1024 * 1024
+    request = urllib.request.Request(url, headers={'Range': f'bytes={have}-'} if have else {})
+    try:
+        stream = opener.open(request, timeout=60)
+    except urllib.error.HTTPError as error:
+        if error.code != 416 or not have:
+            raise
+        # The server holds fewer bytes than the partial file: start over.
+        have = 0
+        stream = opener.open(url, timeout=60)
+    with stream:
+        if have and (stream.status != 206 or not (stream.headers.get('Content-Range') or '').startswith(f'bytes {have}-')):
+            have = 0
+        if have:
+            print(f'Resuming {logical_path} at {have / mib:.1f} MiB of {size / mib:.1f} MiB', flush=True)
+        flags = os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW
         with os.fdopen(os.open(partial, flags, 0o600), 'wb') as output:
-            count, reported = 0, 0
+            output.truncate(have)
+            output.seek(have)
+            count = have
+            reported = count * 10 // size
             window, window_count = time.monotonic(), 0
             while True:
-                block = stream.read1(1024 * 1024)
+                block = stream.read1(mib)
                 if not block:
                     break
                 count += len(block)
                 if count > size:
+                    partial.unlink()
                     raise ArtifactError('Artifact exceeds published size: ' + logical_path)
                 output.write(block)
                 window_count += len(block)
                 if time.monotonic() - window >= SLOW_SECONDS:
                     if window_count < SLOW_BYTES:
                         raise ArtifactError(f'Artifact download stalled (under {SLOW_BYTES // 1024} KiB in {SLOW_SECONDS} s): '
-                                                f'{logical_path}. Check the network, then rerun the command.')
+                                                f'{logical_path}. Check the network, then rerun the command; the download continues where it stopped.')
                     window, window_count = time.monotonic(), 0
-                if size >= 50 * 1024 * 1024 and count * 10 // size > reported:
+                if count * 10 // size > reported:
                     reported = count * 10 // size
                     try:
-                        print(f'Downloading {logical_path}: {reported * 10}% of {size // (1024 * 1024)} MiB', flush=True)
+                        print(f'Downloading {logical_path}: {reported * 10}% of {size / mib:.1f} MiB', flush=True)
                     except BrokenPipeError:
                         # Nobody reads the output any more (the installer's terminal is gone);
                         # stop instead of retrying it as a network error.

deploy/node/node_install.py

@@ -487,15 +487,16 @@ def require_disk(root, args, program_manifest, manifest, names):
             entry = distribution.artifact(program_manifest if name in provider_assets.artifacts(args.provider, ("node",)) else manifest, name)
             target = root / name
             if not (target.is_file() and distribution.matches(target, entry)):
-                needed += entry["size"]
+                needed += entry["size"] - distribution.resumable(distribution.partial_path(target), entry["size"])
     if args.provider == "docker":
         imported = distribution.docker_image(manifest, "runtime", DOCKER) is not None
     else:
         imported = (root / MICRO[1]).is_file() and micro_image_ready(root, args, manifest)
     if not imported:
         archive = distribution.artifact(manifest, "images/runtime.tar.gz")
-        if not (root / "images/runtime.tar.gz").is_file():
-            needed += archive["size"]
+        packed = root / "images/runtime.tar.gz"
+        if not packed.is_file():
+            needed += archive["size"] - distribution.resumable(distribution.partial_path(packed), archive["size"])
         if not (root / "images/runtime.tar").is_file():
             needed += archive["unpacked_size"]
         if args.provider == "microsandbox":