OXYGENHUB

Flow: a job's log says which pinned subrepos it was not given, and why (closes #56) afd0be382257

A pin the run's actor cannot read, or whose repository is not on this
hub, is still left out of the claim, but the job's log now starts with a
flow line naming the path, the repository and the reason, rather than
the job failing later on missing files.
Author
Andy McCall @andymccall <[email protected]>
When
· 2026-10-01 00:56 UTC
Signature
✔ signed by Andy McCall
Hash
afd0be382257ca01f5326008ef73036b1b096ea9abb8d9e44bd0c1b3df305865
Depends on
25091e5c3bba 42fa95540468 fd3db644df22
Depended on by
4ac01a732552
In bookmarks
api-unknown-fields cancel-landed-runs clear-waiting close-proposal-140 diffusion-pin-bookmark draft-ready image-check job-waits main oh-comment-body-file proposal-list-fast run-one-job wiki-says-so
In tags
v0.2.0 v0.3.0 v0.3.1 v0.4.0 v0.4.1 v0.4.2 v0.5.0 v0.5.1 v0.5.2 v0.6.0 v0.6.1 v0.7.0 v0.8.0 v0.9.0 v0.9.1 v0.9.11 v0.9.15 v0.9.16 v0.9.19 v0.9.2 v0.9.20 v0.9.21 v0.9.22 v0.9.23 v0.9.24 v0.9.3 v0.9.4 v0.9.5 v0.9.6 v0.9.7 v0.9.8 v0.9.9
Edits
11 atoms, 0 tree operations, 2 files

Browse the tree at this change

hide whitespace changes
modified crates/hub/src/flow.rs +29 −6
… 
150 unchanged lines
151 151 
        subrepos: vec![],
152 152 
    };
153 153 
    let mut claim = claim;
154 - 
    let (inputs, ids) = subrepo_inputs(&app, &row, &run).await;
154 + 
    let (inputs, ids, left_out) = subrepo_inputs(&app, &row, &run).await;
155 155 
    if !ids.is_empty() {
156 156 
        fdb::set_also_read(&app.db, job.id, &ids).await?;
157 157 
    }
158 + 
    // Say at the top of the job's log what it will not have (oxygenhub#56),
159 + 
    // rather than let it fail later on missing files.
160 + 
    if !left_out.is_empty() {
161 + 
        let records: Vec<flow_core::api::Record> =
162 + 
            left_out.into_iter().map(|text| flow_core::api::Record { t: 0, stream: "flow".into(), text }).collect();
163 + 
        fdb::append_log(&app.db, job.id, &records, MAX_RECORDS as i64).await?;
164 + 
    }
158 165 
    claim.subrepos = inputs;
159 166 
    Ok(Json(claim).into_response())
160 167 
}
… 
36 unchanged lines
197 204 
/// The Oxygen subrepos the run's state pins, and those their states pin in
198 205 
/// turn (parents first, paths joined), that live on this hub and that the
199 206 
/// run's pusher may read: what the job's workspace needs beside its own
200 - 
/// tree. A pin the hub cannot serve is left out, with what it pins; the job
201 - 
/// then fails on the missing files with an ordinary error.
202 - 
async fn subrepo_inputs(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run) -> (Vec<flow_core::api::SubrepoInput>, Vec<i64>) {
203 - 
    let Some(name) = crate::repos::RepoName::new(&row.owner, &row.name) else { return (vec![], vec![]) };
207 + 
/// tree. A pin the hub cannot serve is left out, with what it pins, and
208 + 
/// the third list says why, for the job's log.
209 + 
async fn subrepo_inputs(
210 + 
    app: &Shared,
211 + 
    row: &db::repos::RepoRow,
212 + 
    run: &fdb::Run,
213 + 
) -> (Vec<flow_core::api::SubrepoInput>, Vec<i64>, Vec<String>) {
214 + 
    let Some(name) = crate::repos::RepoName::new(&row.owner, &row.name) else { return (vec![], vec![], vec![]) };
204 215 
    let heads: Vec<oxy_core::hash::ChangeHash> = run.head_list().into_iter().filter_map(oxy_core::hash::ChangeHash::from_hex).collect();
205 216 
    let pusher = db::accounts::account_by_name(&app.db, &run.actor).await.ok().flatten();
206 217 
    let rows = db::repos::list(&app.db).await.unwrap_or_default();
207 218 
    let mut inputs = Vec::new();
208 219 
    let mut ids = Vec::new();
220 + 
    let mut left_out = Vec::new();
221 + 
    let actor = if run.actor.is_empty() { "the run's starter".to_string() } else { run.actor.clone() };
209 222 
    // (path prefix, repository, heads, depth), parents before children.
210 223 
    let mut queue = std::collections::VecDeque::from([(String::new(), name, heads, 0usize)]);
211 224 
    while let Some((prefix, name, heads, depth)) = queue.pop_front() {
… 
2 unchanged lines
214 227 
        }
215 228 
        for (path, id, pinned) in pins_in(app, name, heads).await {
216 229 
            let path = if prefix.is_empty() { path } else { format!("{prefix}/{path}") };
230 + 
            let mut hosted = false;
217 231 
            for other in &rows {
218 232 
                let Some(other_name) = crate::repos::RepoName::new(&other.owner, &other.name) else { continue };
219 233 
                let matches = app.repos.read(&other_name, move |r, _| Ok(r.id().ok() == Some(id))).await.unwrap_or(false);
220 234 
                if !matches {
221 235 
                    continue;
222 236 
                }
237 + 
                hosted = true;
223 238 
                let role = crate::auth::roles::role_for(&app.db, pusher.as_ref(), other).await.ok().flatten();
224 239 
                if crate::auth::roles::allows(role, crate::auth::Access::Read) {
225 240 
                    inputs.push(flow_core::api::SubrepoInput {
… 
6 unchanged lines
232 247 
                        ids.push(other.id);
233 248 
                    }
234 249 
                    queue.push_back((path.clone(), other_name, pinned.clone(), depth + 1));
250 + 
                } else {
251 + 
                    left_out.push(format!(
252 + 
                        "subrepo {path} ({}/{}) is left out: {actor} cannot read it, so the job has no files there",
253 + 
                        other.owner, other.name
254 + 
                    ));
235 255 
                }
236 256 
                break;
257 + 
            }
258 + 
            if !hosted {
259 + 
                left_out.push(format!("subrepo {path} is left out: the repository it pins is not on this hub"));
237 260 
            }
238 261 
        }
239 262 
    }
240 - 
    (inputs, ids)
263 + 
    (inputs, ids, left_out)
241 264 
}
242 265 

    

  
    

      243
      266
       
      
async fn running_job(app: &Shared, id: i64) -> Result<fdb::Job, ApiError> {

    

  
    

      
      
      …
      
98 unchanged lines

    

  
  

  





  
    modified
    crates/hub/tests/flow.rs
    +8 −0
  
  
  

  
    

      
      
      …
      
398 unchanged lines

    

  
    

      399
      399
       
      
    assert_eq!(subs[0]["repo"], "andy/lib");

    

  
    

      400
      400
       
      
    assert_eq!(subs[0]["heads"], serde_json::json!(lib_heads.iter().map(|h| h.to_hex()).collect::<Vec<_>>()));

    

  
    

      401
      401
       
      
    assert!(subs[0]["clone_url"].as_str().unwrap().ends_with("/andy/lib"));

    

  
    

      
      402
      +
      
    // The job's log says vault was left out, and why (oxygenhub#56).

    

  
    

      
      403
      +
      
    let job_id = job["job_id"].as_i64().unwrap();

    

  
    

      
      404
      +
      
    let lines = db::flow::log(&h.app.db, job_id, 0, 100).await.unwrap();

    

  
    

      
      405
      +
      
    assert!(

    

  
    

      
      406
      +
      
        lines.iter().any(|l| l.stream == "flow" && l.text.contains("subrepo deps/vault (andy/vault) is left out")),

    

  
    

      
      407
      +
      
        "{:?}",

    

  
    

      
      408
      +
      
        lines.iter().map(|l| &l.text).collect::<Vec<_>>()

    

  
    

      
      409
      +
      
    );

    

  
    

      402
      410
       
      

    

  
    

      403
      411
       
      
    // The job token reads its repository and the pinned lib, nothing else.

    

  
    

      404
      412
       
      
    let token = job["token"].as_str().unwrap().to_string();

    

  
    

      
      
      …
      
8 unchanged lines