fix(model-install): preserve install tmpdir when a single multi-file part fails

docs/src/content/docs/development/Architecture/model-manager.mdx

@@ -315,18 +315,19 @@ source6 = HFModelSource(repo_id='runwayml/stable-diffusion-v1-5', subfolder='Ora
 source7 = URLModelSource(url='https://civitai.com/api/download/models/63006')       # model located at a URL
 source8 = URLModelSource(url='https://civitai.com/api/download/models/63006', access_token='letmein') # with an access token
 
-for source in [source1, source2, source3, source4, source5, source6, source7]:
-   install_job = installer.install_model(source)
-
-source2job = installer.wait_for_installs(timeout=120)
-for source in sources:
-    job = source2job[source]
- if job.complete:
-  model_config = job.config_out
-  model_key = model_config.key
-  print(f"{source} installed as {model_key}")
- elif job.errored:
-     print(f"{source}: {job.error_type}.\nStack trace:\n{job.error}")
+for source in [source1, source2, source3, source4, source5, source6, source7, source8]:
+    installer.import_model(source)
+
+jobs = installer.wait_for_installs(timeout=120)
+for job in jobs:
+    if job.complete:
+        model_config = job.config_out
+        model_key = model_config.key
+        print(f"{job.source} installed as {model_key}")
+    elif job.errored:
+        print(f"{job.source}: {job.error_type}.\nStack trace:\n{job.error}")
+    elif job.paused:
+        print(f"{job.source} paused; resume or retry its download to finish installation")
 
 ```
 
@@ -336,7 +337,7 @@ Each call to `import_model()` return a `ModelInstallJob` job, an object which tr
 
 If a remote model is requested, the model's files are downloaded in parallel across a multiple set of threads using the download queue. During the download process, the `ModelInstallJob` is updated to provide status and progress information. After the files (if any) are downloaded, the remainder of the installation runs in a single serialized background thread. These are the model probing, file copying, and config record database update steps.
 
-Multiple install jobs can be queued up. You may block until all install jobs are completed (or errored) by calling the `wait_for_installs()` method as shown in the code example. `wait_for_installs()` will return a `dict` that maps the requested source to its job. This object can be interrogated to determine its status. If the job errored out, then the error type and details can be recovered from `job.error_type` and `job.error`.
+Multiple install jobs can be queued up. Call `wait_for_installs()` to wait for pending download and install work to quiesce. Its returned list contains the tracked jobs and may include a recoverably paused, nonterminal job; resume or retry that job before treating installation as complete. If a job errored out, its error type and details are available as `job.error_type` and `job.error`.
 
 The full list of arguments to `import_model()` is as follows:
 
@@ -411,7 +412,7 @@ The `ModelInstallJob` class has the following structure:
 | Attribute | Type | Description |
 |-----------|------|-------------|
 | `id` | `int` | Integer ID for this job |
-| `status` | `InstallStatus` | An enum of [`waiting`, `downloading`, `running`, `completed`, `error` and `cancelled`] |
+| `status` | `InstallStatus` | An enum of [`waiting`, `downloading`, `running`, `paused`, `completed`, `error` and `cancelled`] |
 | `config_in` | `dict` | Overriding configuration values provided by the caller |
 | `config_out` | `AnyModelConfig` | After successful completion, contains the configuration record written to the database |
 | `inplace` | `boolean` | True if the caller asked to install the model in place using its local path |
@@ -422,45 +423,52 @@ The `ModelInstallJob` class has the following structure:
 
 If the `event_bus` argument was provided, events will also be broadcast to the InvokeAI event bus. The events will appear on the bus as an event of type `EventServiceBase.model_event`, a timestamp and the following event names:
 
-##### `model_install_downloading`
+##### `model_install_download_started` and `model_install_download_progress`
 
-For remote models only, `model_install_downloading` events will be issued at regular intervals as the download progresses. The event's payload contains the following keys:
+For remote models, `model_install_download_started` is emitted when downloading begins, and `model_install_download_progress` is emitted as it progresses. Both payloads contain these keys:
 
 | Key | Type | Description |
 |-----|------|-------------|
-|`source`|str|String representation of the requested source|
-|`local_path`|str|String representation of the path to the downloading model (usually a temporary directory)|
+|`id`|int|ID of the install job|
+|`source`|ModelSource|Requested model source|
+|`local_path`|str|Path where the model is downloading to (usually a temporary directory)|
 |`bytes`|int|How many bytes downloaded so far|
 |`total_bytes`|int|Total size of all the files that make up the model|
 |`parts`|List[Dict]|Information on the progress of the individual files that make up the model|
 
-The parts is a list of dictionaries that give information on each of the components pieces of the download. The dictionary's keys are `source`, `local_path`, `bytes` and `total_bytes`, and correspond to the like-named keys in the main event.
+Each item in `parts` contains `url`, `local_path`, `bytes` and `total_bytes`.
 
-Note that downloading events will not be issued for local models, and that downloading events occur _before_ the running event.
+These download events are not issued for local models. They occur before `model_install_started`.
 
-##### `model_install_running`
+##### `model_install_downloads_complete`
 
-`model_install_running` is issued when all the required downloads have completed (if applicable) and the model probing, copying and registration process has now started.
+`model_install_downloads_complete` is emitted after all required downloads finish. Its payload contains `id` and `source`.
 
-The payload will contain the key `source`.
+##### `model_install_paused`
 
-##### `model_install_completed`
+`model_install_paused` is emitted when a remote install is paused, including when a transient file-download error preserves staging for resume or retry. Its payload contains `id` and `source`.
 
-`model_install_completed` is issued once at the end of a successful installation. The payload will contain the keys `source`, `total_bytes` and `key`, where `key` is the ID under which the model has been registered.
+##### `model_install_started`
+
+`model_install_started` is emitted after any download, when model probing, copying and registration begin. Its payload contains `id` and `source`.
+
+##### `model_install_complete`
+
+`model_install_complete` is emitted once an installation succeeds. Its payload contains `id`, `source`, `key`, `total_bytes` and `config`; `key` is the ID under which the model has been registered, and `total_bytes` may be `null` for a local model.
 
 ##### `model_install_error`
 
-`model_install_error` is emitted if the installation process fails for some reason. The payload will contain the keys `source`, `error_type` and `error`. `error_type` is a short message indicating the nature of the error, and `error` is the long traceback to help debug the problem.
+`model_install_error` is emitted if the installation process fails. Its payload contains `id`, `source`, `error_type` and `error`.
 
 ##### `model_install_cancelled`
 
-`model_install_cancelled` is issued if the model installation is cancelled, or if one or more of its files' downloads are cancelled. The payload will contain `source`.
+`model_install_cancelled` is emitted when the installation is cancelled. Its payload contains `id` and `source`.
 
 ##### Following the model status
 
-You may poll the `ModelInstallJob` object returned by `import_model()` to ascertain the state of the install. The job status can be read from the job's `status` attribute, an `InstallStatus` enum which has the enumerated values `WAITING`, `DOWNLOADING`, `RUNNING`, `COMPLETED`, `ERROR` and `CANCELLED`.
+You may poll the `ModelInstallJob` object returned by `import_model()` to ascertain the state of the install. The job status can be read from the job's `status` attribute, an `InstallStatus` enum which has the enumerated values `WAITING`, `DOWNLOADING`, `RUNNING`, `PAUSED`, `COMPLETED`, `ERROR` and `CANCELLED`. A paused remote install can be resumed or its failed files retried while its staging data is retained.
 
-For convenience, install jobs also provided the following boolean properties: `waiting`, `downloading`, `running`, `complete`, `errored` and `cancelled`, as well as `in_terminal_state`. The last will return True if the job is in the complete, errored or cancelled states.
+For convenience, install jobs also provide the following boolean properties: `waiting`, `downloading`, `running`, `paused`, `complete`, `errored` and `cancelled`, as well as `in_terminal_state`. The last will return True if the job is in the complete, errored or cancelled states.
 
 #### Model configuration and probing
 
@@ -486,15 +494,15 @@ This section describes additional methods provided by the installer class.
 
 #### jobs = installer.wait_for_installs([timeout])
 
-Block until all pending installs are completed or errored and then returns a list of completed jobs. The optional `timeout` argument will return from the call if jobs aren't completed in the specified time. An argument of 0 (the default) will block indefinitely.
+Wait until pending download and install work has quiesced, then return the tracked jobs. Transient download failures can leave a job in `PAUSED`; this nonterminal job is returned and does not block the method. Resume or retry it before treating installation as complete. The optional `timeout` argument raises if work does not quiesce in time; 0 (the default) waits indefinitely.
 
 #### jobs = installer.wait_for_job(job, [timeout])
 
-Like `wait_for_installs()`, but block until a specific job has completed or errored, and then return the job.  The optional `timeout` argument will return from the call if the job doesn't complete in the specified time. An argument of 0 (the default) will block indefinitely.
+Block until a specific job reaches a terminal state or pauses, then return it. While a resume or retry of a paused job is being set up, this keeps waiting for it to take effect. A paused job is nonterminal and stays paused until it is resumed or its failed files are retried, so check `job.paused` before treating the install as finished. The optional `timeout` argument raises if the job neither finishes nor pauses in time; 0 (the default) waits indefinitely.
 
 #### jobs = installer.list_jobs()
 
-Return a list of all active and complete `ModelInstallJobs`.
+Return a list of all tracked `ModelInstallJob` objects, including queued, paused, and terminal jobs.
 
 #### jobs = installer.get_job_by_source(source)
 
@@ -629,7 +637,7 @@ The `status` field is updated by the queue to indicate where the job is in its l
 
 In case of an error, the Exception that caused the error will be placed in the `error` field, and the job's status will be set to `DownloadJobStatus.ERROR`.
 
-After an error occurs, any partially downloaded files will be deleted from disk, unless `preserve_partial_downloads` was set to True at job creation time (or set to True any time before the error occurred). Note that since all InvokeAI model install operations involve downloading files to a temporary directory that has a limited lifetime, this flag is not used by the model installer.
+Download errors can leave a `.downloading` file behind; cleanup is owned by the caller. When a part of a model install fails with a transient error (a connection failure or timeout, an interrupted transfer including a dropped TLS connection, HTTP 408, 429 or 5xx, or the `.downloading` file vanishing before its final rename) and the install's staging directory holds downloaded bytes (a non-empty completed file or resumable partial, from this attempt or an earlier one), the installer pauses the install and preserves its staging directory and marker for resume or restart. The failed part keeps its `error_type` and `error` in the job's `download_parts`. Any other error (for example HTTP 401, 403 or 404, a TLS certificate that fails verification, insufficient disk space, or an unsafe filename), or a transient one with no recoverable data, errors the install and removes the temporary staging directory.
 
 There are a series of subclasses of `DownloadJobBase` that provide support for specific types of downloads. These are:
 

invokeai/app/api/routers/model_manager.py

@@ -1333,7 +1333,7 @@ def resume_model_install_job(
         raise HTTPException(status_code=415, detail=str(e))
     try:
         installer.resume_job(job)
-    except (InstallCancellationConflictError, InstallDownloadConflictError) as e:
+    except (InstallCancellationConflictError, InstallDownloadConflictError, TimeoutError) as e:
         raise HTTPException(status_code=409, detail=str(e)) from e
     return job
 

invokeai/app/api/sockets.py

@@ -46,6 +46,7 @@
     ModelInstallDownloadsCompleteEvent,
     ModelInstallDownloadStartedEvent,
     ModelInstallErrorEvent,
+    ModelInstallPausedEvent,
     ModelInstallStartedEvent,
     ModelLoadCompleteEvent,
     ModelLoadStartedEvent,
@@ -126,6 +127,7 @@ class BulkDownloadSubscriptionEvent(BaseModel):
     ModelInstallDownloadStartedEvent,
     ModelInstallDownloadProgressEvent,
     ModelInstallDownloadsCompleteEvent,
+    ModelInstallPausedEvent,
     ModelInstallStartedEvent,
     ModelInstallCompleteEvent,
     ModelInstallCancelledEvent,
@@ -144,6 +146,7 @@ class BulkDownloadSubscriptionEvent(BaseModel):
     ModelInstallDownloadStartedEvent,
     ModelInstallDownloadProgressEvent,
     ModelInstallDownloadsCompleteEvent,
+    ModelInstallPausedEvent,
     ModelInstallStartedEvent,
     ModelInstallCompleteEvent,
     ModelInstallCancelledEvent,

invokeai/app/invocations/flux/flux_redux.py

@@ -21,6 +21,7 @@
 )
 from invokeai.app.invocations.model import ModelIdentifierField
 from invokeai.app.invocations.primitives import ImageField
+from invokeai.app.services.model_install.model_install_common import InstallDownloadConflictError
 from invokeai.app.services.model_records.model_records_base import ModelRecordChanges
 from invokeai.app.services.shared.invocation_context import InvocationContext
 from invokeai.backend.flux.redux.flux_redux_model import FluxReduxModel
@@ -153,10 +154,28 @@ def _get_siglip_model(self, context: InvocationContext) -> AnyModelConfig:
             config_overrides = ModelRecordChanges(name=siglip.name, type=ModelType.SigLIP)
 
             # Queue the job
-            job = context._services.model_manager.install.heuristic_import(siglip.source, config=config_overrides)
-
-            # Wait for up to 10 minutes - model is ~3.5GB
-            context._services.model_manager.install.wait_for_job(job, timeout=600)
+            installer = context._services.model_manager.install
+            job = installer.heuristic_import(siglip.source, config=config_overrides)
+            # An earlier run's install may have paused on a download error; retrying is what this call is for.
+            try:
+                try:
+                    if job.paused:
+                        installer.resume_job(job)
+                except InstallDownloadConflictError:
+                    # Another invocation is already resuming it; wait_for_job() waits for that resume to land.
+                    pass
+                # Wait for up to 10 minutes - model is ~3.5GB
+                installer.wait_for_job(job, timeout=600)
+            except TimeoutError:
+                # Model-specific failure below includes the install state and recovery path.
+                pass
+
+            if not job.complete:
+                raise RuntimeError(
+                    f"Could not install the SigLIP model required by FLUX Redux ({siglip.name}): install status is "
+                    f"'{job.status.value}'{f' ({job.error})' if job.error else ''}. "
+                    "A paused install can be resumed or retried from the Model Manager."
+                )
 
             siglip_models = context.models.search_by_attrs(
                 name=siglip.name,

invokeai/app/invocations/sd/ip_adapter.py

@@ -9,6 +9,7 @@
 from invokeai.app.invocations.model import ModelIdentifierField
 from invokeai.app.invocations.primitives import ImageField
 from invokeai.app.invocations.util import validate_begin_end_step, validate_weights
+from invokeai.app.services.model_install.model_install_common import InstallDownloadConflictError
 from invokeai.app.services.model_records.model_records_base import ModelRecordChanges
 from invokeai.app.services.shared.invocation_context import InvocationContext
 from invokeai.backend.model_manager.configs.factory import AnyModelConfig
@@ -220,7 +221,25 @@ def get_clip_image_encoder(
             job = installer.heuristic_import(
                 image_encoder_model_id, ModelRecordChanges(name=image_encoder_model_name, type=ModelType.CLIPVision)
             )
-            installer.wait_for_job(job, timeout=600)  # Wait for up to 10 minutes
+            # An earlier run's install may have paused on a download error; retrying is what this call is for.
+            try:
+                try:
+                    if job.paused:
+                        installer.resume_job(job)
+                except InstallDownloadConflictError:
+                    # Another invocation is already resuming it; wait_for_job() waits for that resume to land.
+                    pass
+                installer.wait_for_job(job, timeout=600)  # Wait for up to 10 minutes
+            except TimeoutError:
+                # Model-specific failure below includes the install state and recovery path.
+                pass
+
+            if not job.complete:
+                raise RuntimeError(
+                    f"Could not install the image encoder required by this IP Adapter ({image_encoder_model_name}): "
+                    f"install status is '{job.status.value}'{f' ({job.error})' if job.error else ''}. "
+                    "A paused install can be resumed or retried from the Model Manager."
+                )
             image_encoder_models = context.models.search_by_attrs(
                 name=image_encoder_model_name, base=BaseModelType.Any, type=ModelType.CLIPVision
             )