feat(depot): release completed bundle downloads during exports

CHANGELOG.md

@@ -9,6 +9,7 @@ schemas in `nnnotes.schemas` and the documented Python API. Before 1.0 a minor v
 ### Features
 
 - Store content, task inputs, results and catalog versions in a persistent depot with optional remote storage, verified collection and bounded source downloads.
+- Release recoverable bundle downloads after successful conversion batches under the configured budget and command leases.
 - Assemble Live2D, room, spot, live, story and UI products from shared object tables and explicit manifest references.
 - Generate localized live and story text products with game or supplied fonts, shared glyph atlases and material records.
 - Build named sites through depot product, player read-set, asset and index tasks.

docs/architecture.md

@@ -48,6 +48,11 @@ The site's `Builder` composes product snapshots, catalog indexes and player page
 
 Converters use bounded process-local memoization for repeated reads and computations. Persistent reuse is keyed by depot tasks and their content inputs. Clearing a process memo releases its values; a later task can reuse its committed depot result.
 
+The coordinator observes successful object batches after worker readers close. Their recoverable bundle downloads
+can then be released toward the configured budget under the command's exclusive lease. Preliminary episode
+requests retain inputs until the expanded product request is known. Stored conversion results remain available
+for assembly and layouts.
+
 `export episode`, `export shader`, `export audio` and `music-data` expose subject-specific data and conversions through the Session. All commands use the readers and converters below the pipeline layer.
 
 ## Outputs

docs/commands.md

@@ -69,6 +69,14 @@ Catalog files are kept. A removed file is downloaded again when a command needs
 identity (sha256 and size) of every bundle and raw file it reads, so `export assets` plans with files whose name
 carries the catalog's hash without downloading them again.
 
+During product and asset requests, a completed object-conversion batch makes its bundle downloads eligible for
+release toward the same budget. Readers finish before release; active batches, failed batches and preliminary
+episode requests stay protected. A batch already present in the depot can release its downloaded input as well.
+Only files recoverable through an explicit catalog or depot locator are released, and their contents are checked
+against the input identity. `all` retains downloads throughout the run; `0` releases each eligible batch without
+scanning the download directory. With a finite size, finished files are evicted in LRU order and active files can
+temporarily exceed the budget. Later requests can download a released bundle again.
+
 Commands may run at the same time on one depot: no command removes downloads while another one holds a lease, and a
 command that starts while one trims waits for it. The lease of a process that ended without releasing it is dropped
 by the next command; a lease written on another host counts for a day.

docs/configuration.md

@@ -110,7 +110,7 @@ exist when they are given.
 | `[paths] player` | `NNNOTES_PATHS_PLAYER` | `--player` (of `site build`) | ournotes-player checkout (built) or installed package |
 | `[paths] fonts.<language>` | `NNNOTES_PATHS_FONTS_<LANGUAGE>` (e.g. `NNNOTES_PATHS_FONTS_ZH_HANT`) | `--font <language>=PATH` (of `site build`) | font file (OpenType or TrueType) the story text of a language is drawn with (`--fonts open`); also read for the other story languages whose game fonts fall back to that language's font |
 | `[paths] fonts.emoji` | `NNNOTES_PATHS_FONTS_EMOJI` | `--font emoji=PATH` (of `site build`) | colour emoji font (PNG bitmap glyphs: CBDT or sbix) the stories' emoji sprites are drawn from (`--fonts open`); Noto Color Emoji (SIL Open Font License 1.1) is the tested font. Optional: without it the sprites keep their layout with empty glyphs, listed in `ui/fonts.json` `coverage.sprites.missing` |
-| `[depot] downloads` | `NNNOTES_DEPOT_DOWNLOADS` | — | how much of the depot's downloaded bundles and raw files to keep once no command uses the depot: a size (`500MB`, `20GiB`; units B, KB, MB, GB, TB and KiB, MiB, GiB, TiB), `all`, or `0` (unset) ([commands.md](commands.md#downloads)) |
+| `[depot] downloads` | `NNNOTES_DEPOT_DOWNLOADS` | — | download-cache budget, applied to finished conversion batches and when the last command ends: a size (`500MB`, `20GiB`; units B, KB, MB, GB, TB and KiB, MiB, GiB, TiB), `all`, or `0` (unset) ([commands.md](commands.md#downloads)) |
 | `[depot] remote` | `NNNOTES_DEPOT_REMOTE` | — | a remote of the depot shared by the machines that use it: `s3://<bucket>/<prefix>` (S3-compatible object storage; `pip install 'nnnotes[remote]'`) or `file:///<directory>` ([contracts.md](contracts.md#depot-remote)) |
 | `[depot] endpoint` | `NNNOTES_DEPOT_ENDPOINT` | — | endpoint URL of the S3-compatible object storage; unset: AWS S3 |
 | `[depot] access_key_id` | `NNNOTES_DEPOT_ACCESS_KEY_ID` | — | access key id of the object storage; unset: boto3 looks for credentials itself (`AWS_ACCESS_KEY_ID`, `~/.aws`) |
@@ -188,9 +188,10 @@ story text); it is not a setting.
 | `catalog versions` / `import` / `diff`, `depot verify` | the depot; `import` reads the APK's catalog from `[paths] apk` when it is set |
 | `catalog fetch` | the depot, `[catalog] region` and `language`, that region's `cdn`; its `api` (optional) for the resource version label |
 
-The region, its `cdn`, `[bundle] key` and `nonce_seed` are read only when a file must be downloaded: when the
-catalog and every bundle a command needs are downloaded, none of them is needed. A download that needs a missing
-one stops the command with one line naming the file and the setting (see Errors).
+The region, its `cdn`, `[bundle] key` and `nonce_seed` are read only when a file must be downloaded. A product
+request can read a downloaded catalog and its bundle files without these settings. Use `[depot] downloads = all`
+for an offline sequence of requests; a bounded budget can release a bundle that a later request needs again.
+A download that needs a missing setting stops the command with one line naming the file and setting (see Errors).
 
 ## Errors
 

src/nnnotes/cli/assets.py

@@ -1317,8 +1317,10 @@ def cmd_export(args, cfg):
 
 def request_orchestrator(ws, cfg):
     """The shared executor for product requests."""
+    from ..pipeline.downloads import BundleDownloads
     from ..pipeline.orchestrate import Orchestrator
 
+    lease, budget = getattr(ws.args, "lease", None), Session(cfg).downloads_budget()
     return Orchestrator(
         ws.depot,
         ws.scheduled,
@@ -1329,12 +1331,14 @@ def request_orchestrator(ws, cfg):
         initializer=_worker_init,
         initargs=(cfg,),
         log=ws.log,
+        on_batch_done=BundleDownloads(ws.depot, lease, budget) if lease is not None and budget is not None else None,
     )
 
 
 def run_request(args, cfg, ws: Workspace, expand=None, prepared=None):
     """Execute a product request and optional expansion without materializing an output layout."""
     from ..pipeline import builtins
+    from ..pipeline.downloads import BundleDownloads
     from ..pipeline.orchestrate import failure
 
     if prepared is None:
@@ -1343,6 +1347,8 @@ def run_request(args, cfg, ws: Workspace, expand=None, prepared=None):
         orch = request_orchestrator(ws, cfg)
     else:
         pipeline, params, orch = prepared
+    if expand is not None:
+        orch.on_batch_done = None  # the episode supplies the roots of the complete product request
     ws.run_index()
     facts = ws.facts(pipeline, fetch=True, background=True)
     sel = ws.selected
@@ -1374,6 +1380,10 @@ def run_request(args, cfg, ws: Workspace, expand=None, prepared=None):
         aborted = aborted or str(e)
     if isinstance(facts["raw"], Later) and aborted is None:
         ws.log(f"{sum(p['kind'] == 'raw' for p in ws.problems())} raw files not readable")
+    released = orch.on_batch_done
+    if isinstance(released, BundleDownloads) and released.removed:
+        run.log.append({"event": "downloads.released", "files": released.removed, "bytes": released.bytes_removed})
+        ws.log(f"downloads: released {released.removed} bundle files ({released.bytes_removed} bytes)")
     if expand is not None and aborted is None:
         more = expand(run)
         if more is not None:

src/nnnotes/core/downloads.py

@@ -1,11 +1,12 @@
-"""The downloads of a depot as a cache bounded in size: what a command reads again stays while commands run, and the
-bundles and raw files are trimmed to the budget when the last command using the depot ends.
+"""Budgeted depot downloads, with finished-batch release and command leases protecting active readers.
 
     <depot>/downloads/                the game files as downloaded (source.catalog)
     <depot>/leases/<lease>.json       one per running command: {"host", "pid", "started"}
     <depot>/leases/.exclusive         held by a command that must be the only one (trimming, `depot gc`)
 
-A command holds a lease while it runs (`lease`). When it ends and no other lease is live (a lease of this host whose
+A command holds a lease while it runs (`lease`). A successful object conversion batch may release its recoverable
+bundle downloads toward the budget (`release`) while it is the only command. When the command ends and no other
+lease is live (a lease of this host whose
 process runs, or of another host younger than a day), the files under the `bundles/` and `raw/` directories of
 downloads/ are removed, least recently used first (a file's modification time: the catalog sets it when it hands
 the file out), until they fit the budget. Catalog files stay. A removed file is downloaded again when a command needs
@@ -25,12 +26,12 @@
 import secrets
 import socket
 import time
-from collections.abc import Generator, Iterator
+from collections.abc import Generator, Iterable, Iterator
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Any
 
-__all__ = ["evictable", "exclusive", "lease", "parse_budget", "trim"]
+__all__ = ["evictable", "exclusive", "lease", "parse_budget", "release", "trim"]
 
 FOREIGN_LEASE_SECONDS = 24 * 3600  # a lease of another host counts as live this long
 EXCLUSIVE = ".exclusive"
@@ -71,39 +72,79 @@ def evictable(downloads: Path) -> Iterator[Path]:
                 yield p
 
 
-def trim(downloads: Path, budget: int | None, *, dry_run: bool = False) -> dict[str, int]:
-    """Remove the least recently used bundles and raw files until the rest fit `budget` bytes (None: nothing).
-    -> {"kept", "removed", "bytesKept", "bytesRemoved"}."""
-    files: list[tuple[int, int, Path]] = []
-    for p in evictable(downloads):
+def _files(paths: Iterable[Path]) -> list[tuple[int, int, Path]]:
+    out: list[tuple[int, int, Path]] = []
+    for path in paths:
         try:
-            st = p.stat()
+            stat = path.stat()
         except OSError:
             continue
-        files.append((st.st_mtime_ns, st.st_size, p))
+        out.append((stat.st_mtime_ns, stat.st_size, path))
+    return out
+
+
+def _remove(files: list[tuple[int, int, Path]], needed: int, dry_run: bool) -> tuple[list[Path], int]:
+    removed: list[Path] = []
+    removed_bytes = 0
+    for _, size, path in sorted(files, key=lambda f: (f[0], str(f[2]))):
+        if removed_bytes >= needed:
+            break
+        if not dry_run:
+            try:
+                path.unlink()
+            except OSError:
+                continue
+        removed.append(path)
+        removed_bytes += size
+    return removed, removed_bytes
+
+
+def trim(downloads: Path, budget: int | None, *, dry_run: bool = False) -> dict[str, int]:
+    """Remove the least recently used bundles and raw files until the rest fit `budget` bytes (None: nothing).
+    -> {"kept", "removed", "bytesKept", "bytesRemoved"}."""
+    files = _files(evictable(downloads))
     total = sum(f[1] for f in files)
-    removed = removed_bytes = 0
-    if budget is not None:
-        for _, size, p in sorted(files, key=lambda f: (f[0], str(f[2]))):
-            if total - removed_bytes <= budget:
-                break
-            if not dry_run:
-                try:
-                    p.unlink()
-                except OSError:  # open elsewhere (Windows): kept
-                    continue
-            removed += 1
-            removed_bytes += size
-        if not dry_run and removed:
-            _prune_dirs(downloads)
+    removed, removed_bytes = _remove(files, 0 if budget is None else total - budget, dry_run)
+    if not dry_run and removed:
+        _prune_dirs(downloads)
     return {
-        "kept": len(files) - removed,
-        "removed": removed,
+        "kept": len(files) - len(removed),
+        "removed": len(removed),
         "bytesKept": total - removed_bytes,
         "bytesRemoved": removed_bytes,
     }
 
 
+def release(downloads: Path, ready: Iterable[Path], budget: int | None) -> dict[str, int]:
+    """Trim only finished downloads, in LRU order, toward the cache budget. The caller holds an exclusive lease.
+
+    Active files remain protected even when they exceed the budget. A zero budget examines only finished files.
+    Paths must resolve below a bundles/ or raw/ directory within downloads; partial files are retained.
+    """
+    if budget is None:
+        return {"removed": 0, "bytesRemoved": 0}
+    root = downloads.resolve()
+    paths: set[Path] = set()
+    for path in ready:
+        path = path.resolve()
+        if not path.is_relative_to(root) or not {"bundles", "raw"} & set(path.relative_to(root).parts[:-1]):
+            raise ValueError("released file must be a bundle or raw file within the downloads directory")
+        if path.is_file() and not path.name.endswith(".part"):
+            paths.add(path)
+    files = _files(paths)
+    total = sum(f[1] for f in files) if budget == 0 else sum(f[1] for f in _files(evictable(root)))
+    removed, size = _remove(files, total - budget, False)
+    for path in removed:
+        parent = path.parent
+        while parent != root:
+            try:
+                parent.rmdir()
+            except OSError:
+                break
+            parent = parent.parent
+    return {"removed": len(removed), "bytesRemoved": size}
+
+
 def _prune_dirs(root: Path) -> None:
     for d in sorted((p for p in root.rglob("*") if p.is_dir()), key=lambda p: len(p.parts), reverse=True):
         try:

src/nnnotes/pipeline/downloads.py

@@ -0,0 +1,56 @@
+"""Budgeted release of recoverable bundle downloads after complete conversion batches."""
+
+from pathlib import Path
+
+from ..core import downloads
+from .contract import Input
+
+
+class BundleDownloads:
+    """A main-process batch observer; only closed, recoverable bundle inputs become eligible for release."""
+
+    def __init__(self, depot, lease: Path | None, budget: int | None):
+        self.depot, self.lease, self.budget = depot, Path(lease) if lease is not None else None, budget
+        self.ready: dict[Path, Input] = {}
+        self.removed = self.bytes_removed = 0
+
+    def __call__(self, tasks):
+        if self.budget is None or self.lease is None or not self.lease.is_file():
+            return
+        if self.lease.parent.resolve() != (self.depot.root / "leases").resolve() or self.lease.suffix != ".json":
+            return
+        if not tasks or any(task.stage != "unity.export" for task in tasks):
+            return
+        root = self.depot.downloads.resolve()
+        for task in tasks:
+            inp = task.input("bundle")
+            kinds = {loc["kind"] for loc in inp.locators}
+            if "catalog" not in kinds and not ("depot" in kinds and self.depot.has(inp.sha256, inp.size)):
+                continue
+            for loc in inp.locators:
+                if loc["kind"] != "download":
+                    continue
+                path = (root / loc["path"]).resolve()
+                if path.is_relative_to(root) and "bundles" in path.relative_to(root).parts[:-1]:
+                    self.ready[path] = inp
+        if not self.ready:
+            return
+        with downloads.exclusive(self.depot.root, self.lease) as alone:
+            if not alone:
+                return
+            valid = []
+            for path, inp in list(self.ready.items()):
+                try:
+                    matches = path.is_file() and self.depot.identify(path) == (inp.sha256, inp.size)
+                except OSError:
+                    matches = False
+                if matches:
+                    valid.append(path)
+                else:
+                    self.ready.pop(path)
+            result = downloads.release(root, valid, self.budget)
+            self.removed += result["removed"]
+            self.bytes_removed += result["bytesRemoved"]
+            for path in valid:
+                if not path.exists():
+                    self.ready.pop(path, None)