Image push: a big image's manifest is recorded even when the sender gives up, and a kept one is not read again (closes #341) d9d0d314fdcb

- **A big image's push no longer fails at the last step (#341).** The hub reads a whole image before it accepts the manifest, which on a busy disk outlasted `flow rootfs push`'s wait. The image could then be kept but never offered, and a retry read it all again. The hub now finishes and records the image even if the sender gives up, and a manifest it already keeps is recorded without reading the image again.

    // In a task of its own (#341): reading a big image can outlast the

    // sender's wait, and a sender that gives up must not leave an image

    // kept but never offered.

    let m = tokio::spawn(async move {

        let store = app.images();

        // A manifest kept before was verified then; only its row may be missing.

        let m = match store.manifest(&m.hash).filter(|k| k.chunks == m.chunks) {

            Some(kept) => kept,

            // Kept with the architecture its own files name (oxygenhub#173).

            None => tokio::task::spawn_blocking(move || store.keep_reading_arch(&m))

                .await

                .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?

                .map_err(|e| ApiError(StatusCode::CONFLICT, e))?,

        };

        crate::images::record(&app.db, &m, &who).await?;

        crate::db::accounts::audit(&app.db, None, "flow.image", &who, &format!("{} ({} MiB)", &m.hash[..12], m.size >> 20)).await.ok();

        Ok::<_, ApiError>(m)

    })

    .await

    .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))??;

    let (st, body) = call(&h.app, "PUT", &format!("/api/v1/flow/images/{hash}"), Some("site"), Some(manifest.clone())).await;

    assert_eq!(st, StatusCode::CREATED, "{body}");

    // #341: a sender that gave up during the check left the manifest kept

    // but no row; sending it again records it without reading the image

    // again (a chunk gone from disk would fail a second check).

    sqlx::query("DELETE FROM jet_images").execute(&h.app.db).await.unwrap();

    let gone = h.app.config.data.join("images/chunks").join(&missing[0][..2]).join(&missing[0]);

    let kept = std::fs::read(&gone).unwrap();

    std::fs::remove_file(&gone).unwrap();

    assert!(oxygenhub::images::held(&h.app.db).await.unwrap().contains(&hash));

    std::fs::write(&gone, kept).unwrap();